注: 以下の翻訳の正確性は検証されていません。AIPを利用して英語版の原文から機械的に翻訳されたものです。

トランスフォーム

Foundry の Transforms は、データ処理パイプラインの中核となる構成要素です。入力から出力へのデータの流れを定義し、計算ロジックとリソース要件の両方を1つの宣言的な単位にまとめます。

トランスフォームは Pipelines に登録されます。Pipelines は複数のトランスフォーム定義で構成できます。

トランスフォームを定義する

すべてのトランスフォームは、次の3つの主要な要素で構成されます。

デコレーター： 入力、出力、設定を定義します
関数シグネチャ： デコレーターで宣言された入力と出力に対応します
計算ロジック： 実際のデータ処理コードです

以下は、トランスフォームを定義する際の基本的なパターンです。

Copied!1
2
3
4
5
6
7
8
9
from transforms.api import transform, Input, Output


@transform.using( # デコレーター
    hair_eye_color=Input('/examples/students_hair_eye_color'), # 入力
    processed=Output('/examples/hair_eye_color_processed') # 出力
)
def filter_hair_color(hair_eye_color, processed):  # 関数シグネチャは入力/出力と一致します
    ... # 計算ロジック

Python トランスフォームは、シングルノード（軽量）またはマルチノード（Spark）の計算エンジンで設定できます。デフォルトでは、トランスフォームはシングルノードで実行され、データを Polars ↗ または pandas ↗ の DataFrame として読み込むことができます。Polars は、表形式のデータを加工するための DataFrame ライブラリです。パフォーマンス、安定性、使いやすさで知られています。pandas は、広く採用されている使いやすいデータ分析ツールです。

DataFrame を取得するには、以下に示すように入力に対して適切なメソッドを呼び出します。

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
import polars as pl
from transforms.api import transform, Input, Output


@transform.using(
    hair_eye_color=Input('/examples/students_hair_eye_color'),
    processed=Output('/examples/hair_eye_color_processed')
)
def filter_hair_color(hair_eye_color, processed):
    # データを Polars DataFrame として読み込みます
    df = hair_eye_color.polars()

    # 絞り込みロジックを適用します
    filtered = df.filter(pl.col('hair') == 'Brown')

    # 結果を書き込みます
    processed.write_table(filtered)

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from transforms.api import transform, Input, Output


@transform.using(
    hair_eye_color=Input('/examples/students_hair_eye_color'),
    processed=Output('/examples/hair_eye_color_processed')
)
def filter_hair_color(ctx, hair_eye_color, processed):
    # DuckDB 接続を取得します
    conn = ctx.duckdb().conn

    # SQL を使用して絞り込みロジックを適用します
    filtered = conn.sql("SELECT * FROM hair_eye_color WHERE hair = 'Brown'")

    # 結果を書き込みます
    processed.write_table(filtered)

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from transforms.api import transform, Input, Output


@transform.using(
    hair_eye_color=Input('/examples/students_hair_eye_color'),
    processed=Output('/examples/hair_eye_color_processed')
)
def filter_hair_color(hair_eye_color, processed):
    # データを Pandas DataFrame として読み込みます
    df = hair_eye_color.pandas()

    # 絞り込みロジックを適用します
    filtered = df[df['hair'] == 'Brown']

    # 結果を書き込みます
    processed.write_table(filtered)

以下の例は、トランスフォームが複数の入力と出力を受け取れることを示しています。

Copied!1
2
3
4
5
6
7
8
9
10
11
from transforms.api import transform, Input, Output


@transform.using(
    hair_color=Input('/examples/students_hair_color'),
    eye_color=Input('/examples/students_eye_color'),
    males=Output('/examples/hair_eye_color_males'),
    females=Output('/examples/hair_eye_color_females'),
)
def filter_hair_color(hair_color, eye_color, males, females):
    ...

軽量トランスフォームの構文の選択肢

シングルノード（軽量）トランスフォームは、次のいずれかの構文で定義できます。

@transform.using(...)：推奨 API
@transform.lightweight(...)：軽量トランスフォームを明示的に指定するエイリアス
@transform(...) とともに使用する @lightweight デコレーター：レガシー API

バージョン要件

@transform.using API には transforms バージョン 3.68.0 以降が必要です。Code Repositories で利用可能にするには、リポジトリのアップグレードガイドに従ってリポジトリをアップグレードしてください。transformsLangPythonPluginVersion が 1.978.0 以降であることを確認してください。

Spark API の @transform.spark.using には transforms バージョン 3.95.0 以降が必要です。transformsLangPythonPluginVersion が 1.1003.0 以降であることを確認してください。

transforms のバージョンについて詳しくは、transforms のバージョンの概要を参照してください。

@transform.using のバージョンエラーを解決する

次のようなエラーメッセージが表示される場合、更新された構文に対応していない古いバージョンの transforms ライブラリを使用している可能性があります。

Usage of transform.using() on outdated repository
A function object does not have an attribute using

この問題を解決するには、次の方法を使用します。

リポジトリのアップグレードガイドに従って、リポジトリで使用する transforms ライブラリとリポジトリテンプレートを最新バージョンにアップグレードします。
または、@lightweight と @transform(...) を使用するレガシー構文に切り替えます。

さまざまなデータ形式を扱う

軽量トランスフォーム API は、以下の例に示すように、さまざまな形式でのデータの読み込みをサポートしています。

Copied!1
2
3
4
5
6
7
8
9
10
11
12
@transform.using(output=Output('/Project/folder/output'), dataset=Input('/Project/folder/input'))
def compute(ctx, output, dataset):
    polars_df = dataset.polars()         # polars_df は polars.DataFrame オブジェクトです
    lazy_df = dataset.polars(lazy=True)  # ストリーミングモードを有効にします。lazy_df は polars.LazyFrame です
    pandas_df = dataset.pandas()         # pandas_df は pandas.DataFrame です
    arrow_table = dataset.arrow()        # arrow_table は pyarrow.Table です

    # 入力データセットが事前登録された DuckDB 接続
    conn = ctx.duckdb().conn
    df_from_duckdb = conn.sql("SELECT * FROM dataset").fetchdf()  # dataset は自動的に登録されます

    output.write_table(lazy_df)          # 上記のどの形式も write_table に渡せます

dataset.pandas() を呼び出すには、環境に pandas がインストールされている必要があります。同様に、dataset.polars(...) には Polars が利用可能である必要があります。

計算リソースを設定する

.with_resources() メソッドを使用して、トランスフォームに割り当てる計算リソースを制御できます。

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
from transforms.api import transform, Input, Output


@transform.using(
    large_dataset=Input('/datasets/large_file'),
    processed=Output('/datasets/processed')
).with_resources(
    cpu_cores=8,          # 8個の CPU コアを使用します
    memory_gb=16,         # 16 GB のメモリーを割り当てます
    gpu_type="NVIDIA_T4"
)
def process_large_dataset(large_dataset, processed):
    ...

cpu_cores： CPU コア数です。0.5 などの小数を指定できます。デフォルトは 2 です。
memory_mb または memory_gb： メモリーの割り当てです。両方ではなく、どちらか一方のみを使用します。
gpu_type： トランスフォームに割り当てる GPU の種類です。文字列として指定します。

memory_gb と memory_mb のうち、設定できるのは一方のみです。

デフォルトで許可される最大値は、8コアおよび64 GB のメモリーです。これらの上限を引き上げるには、Palantir サポートにお問い合わせください。

利用可能な GPU の種類と GPU のプロビジョニング要件の詳細については、リソース設定のドキュメントを参照してください。

トランスフォームのコンテキスト

データの変換が、入力データセット以外のものに依存する場合があります。たとえば、変換処理で現在の Spark セッションにアクセスしたり、外部サービスと通信したりする必要がある場合です。このような場合は、変換処理に TransformContext オブジェクトを注入できます。

TransformContext オブジェクトを注入するには、計算関数が ctx という名前のパラメーターを受け取る必要があります。そのため、入力や出力に ctx という名前を付けることはできません。TransformContext のすべての機能については、API リファレンス ↗を参照してください。

Copied!1
2
3
4
5
6
7
8
from transforms.api import transform, Output


@transform.using(
    out=Output('/examples/context')
)
def generate_dataframe(ctx, out):
    ...

PySpark トランスフォーム

デフォルトでは、トランスフォームは単一の計算ノードで実行され、pandas や Polars などのデータ処理ライブラリを利用します。ビッグデータの計算ワークフローでは、Python トランスフォームで PySpark ↗ を利用できます。

PySpark は、Apache Spark バックエンドとやり取りしてデータを高速に処理できるラッパー言語です。Spark は、分散サーバーネットワーク上で非常に大きなデータセットを処理できます。適切に使用すれば、スケーラビリティと信頼性の面で利点があります。ただし、オーバーヘッドとリソース使用量も大きくなるため、小規模から中規模のデータには適していません。

ここでは PySpark トランスフォーム API の概要を説明します。詳細については、Python トランスフォーム（Spark）のドキュメントを参照してください。

PySpark トランスフォームを定義する

PySpark トランスフォームは、標準のトランスフォームと同様の構文で定義します。

Copied!1
2
3
4
5
6
7
8
9
from transforms.api import transform, Input, Output


@transform.spark.using(
    hair_eye_color=Input('/examples/students_hair_eye_color'),
    processed=Output('/examples/hair_eye_color_processed')
)
def filter_hair_color(hair_eye_color, processed):
    ... # ここにトランスフォームのロジックを記述します

Polars や pandas のトランスフォームと同様に、トランスフォームの入力から DataFrame オブジェクトを取得できます。

トランスフォームの入力が返す DataFrame ↗ オブジェクトは、PySpark DataFrame です。PySpark の使用方法について詳しくは、Spark Python API ドキュメント ↗を参照してください。

Copied!1
2
3
4
5
6
7
8
9
10
from transforms.api import transform, Input, Output


@transform.spark.using(
    hair_eye_color=Input('/examples/students_hair_eye_color'),
    processed=Output('/examples/hair_eye_color_processed')
)
def filter_hair_color(hair_eye_color, processed):
    filtered_df = hair_eye_color.dataframe().filter(hair_eye_color.dataframe().hair == 'Brown')
    processed.write_dataframe(filtered_df)

PySpark トランスフォームも、複数の入力と出力を受け取ることができます。

Copied!1
2
3
4
5
6
7
8
9
10
11
from transforms.api import transform, Input, Output


@transform.spark.using(
    hair_color=Input('/examples/students_hair_color'),
    eye_color=Input('/examples/students_eye_color'),
    males=Output('/examples/hair_eye_color_males'),
    females=Output('/examples/hair_eye_color_females'),
)
def filter_hair_color(hair_color, eye_color, males, females):
    ...

PySpark トランスフォームがサポートするデータ処理ライブラリは1つのみなので、利便性のために transform_df デコレーターを使用して DataFrame ↗ に直接アクセスできます。

Copied!1
2
3
4
5
6
7
8
9
from transforms.api import transform_df, Input, Output


@transform_df(
    Output('/examples/hair_eye_color_processed'),
    hair_eye_color=Input('/examples/students_hair_eye_color')
)
def filter_hair_color(hair_eye_color):
    return hair_eye_color.filter(hair_eye_color.hair == 'Brown')

PySpark トランスフォームを設定する

PySpark トランスフォームでは、Spark プロファイルを使用してリソースを設定します。これにより、大規模な計算リソースの使用に対して厳格なアクセス制御を適用できます。

Iceberg テーブル

transforms.tables ライブラリを使用すると、Foundry の Iceberg テーブルを Python トランスフォームの入力および出力として使用できます。同じ3つの要素からなるトランスフォーム構造が適用されます。デコレーターは TableInput と TableOutput を使用して入力と出力を定義します。関数シグネチャは、IcebergInput や IcebergOutput などの実行時ハンドルを受け取ります。計算ロジックは標準の DataFrame コードを使用してデータを読み書きします。

トランスフォームでの Iceberg テーブルの使用方法について詳しくは、Iceberg を使用したトランスフォームを参照してください。

トランスフォームのロジックレベルのバージョン管理

TLLV にはモジュールレベルのインポートが必要であり、モジュールへのパッチ適用はできません

TLLV が正しく機能するには、コード内のすべてのインポートをモジュールレベルで宣言する必要があります。また、別のモジュール内のオブジェクトにパッチを適用したり、その他の方法で変更したりしないでください。プロジェクトがこれらの条件を満たさない場合は、TLLV を無効にする必要があります。詳しくは、以下のコード例を参照してください。

TLLV はデフォルトで有効になっています。 TLLV を無効にするには、transformsPython 設定の tllv を false に設定します。この設定は、Python トランスフォームのサブプロジェクト内の build.gradle ファイルにあります。

transformsPython {
    tllv false
}

トランスフォームのバージョンは、ロジックが古くなっているかどうかを判断する際に、トランスフォームの2つのバージョンを比較するために使用される文字列です。入力に変更がなく、トランスフォームのバージョンにも変更がない場合、トランスフォームの出力は最新です。バージョンが変更されると、トランスフォームの出力は無効化され、再計算されます。

デフォルトでは、トランスフォームのバージョンには以下が含まれます。

トランスフォームが定義されているモジュール
トランスフォームが依存するすべてのモジュール
プロジェクトの依存関係すべて

これらのいずれかが変更されると、バージョン文字列が変更されます。上記の項目に含まれないファイルで変更が発生した場合にも出力を無効化するには、transformsPython 設定で tllvFiles を設定します。たとえば、ファイルの設定を読み込んでいて、その設定が変更されたときに出力を無効化したい場合に使用できます。

トランザクション性

トランスフォームは、出力に対して開始されるトランザクションのコンテキストで実行され、このトランザクションはジョブが成功するとコミットされます。出力にロジックが定義されていない場合、または書き込み中のエラーが捕捉されて無視された場合、トランスフォームが成功すると、その出力に空のスナップショットが作成されます。

←

PREVIOUSPython バージョンのサポート

NEXTパイプライン

→