Python トランスフォーム API

Python トランスフォーム API は、Pipeline を構築するためのクラスとデコレーターを提供します。このページには、利用可能な関数に関する情報が含まれています。クラスについても詳しく確認できます。

---

configure

transforms.api.configure(profile=None, allowed_run_duration=None, run_as_user=False)

• トランスフォームの設定を変更するデコレーターです。
• configure デコレーターは、Transform をラップするために使用する必要があります。

Copied!
1
2
3
4
>>> @configure(profile=['EXECUTOR_MEMORY_MEDIUM'])
... @transform(...)
... def my_compute_function(...):
...     pass

パラメーター

• profile (str ↗ または List[str ↗]、任意)
  • 使用するトランスフォームプロファイルの名前です。
• allowed_run_duration (timedelta ↗、任意)
  • このジョブが失敗するまでに実行できる時間の上限です。このパラメーターは慎重に使用してください。許容時間を設定する際は、データの規模や形式の変化などの変動要因を考慮してください。

• run_as_user (Boolean、任意)
  • トランスフォームをユーザー権限で実行するかどうかを決定します。有効にすると、ジョブを実行するユーザーの権限によって、ジョブの動作が異なる場合があります。

発生する例外

• TypeError ↗
  • ラップしたオブジェクトが Transform オブジェクトでない場合に発生します。

---

incremental

transforms.api.incremental(require_incremental=False, semantic_version=1, snapshot_inputs=None)

• 入力と出力を、それぞれに対応する transforms.api.incremental のオブジェクトに変換するデコレーターです。
• incremental デコレーターは、Transform をラップするために使用する必要があります。

Copied!
1
2
3
4
>>> @incremental()
... @transform(...)
... def my_compute_function(...):
...     pass

このデコレーターは、出力データセットからビルド履歴を読み取り、前回のビルド時点における入力の状態を特定します。この情報を使用して、TransformInput、TransformOutput、および TransformContext オブジェクトを、それぞれに対応するインクリメンタル（差分処理）オブジェクトである IncrementalTransformInput、IncrementalTransformOutput、および IncrementalTransformContext に変換します。

このデコレーターは、transform_df() および transform_pandas() デコレーターをラップするためにも使用できます。これらのデコレーターは、入力に対して引数なしで dataframe() および pandas() を呼び出し、PySpark および pandas の DataFrame オブジェクトを抽出します。つまり、使用される読み取りモードは常に added であり、書き込みモードは incremental デコレーターによって決まります。PySpark または pandas のトランスフォームで追加される出力行が、追加された入力行のみによって決まる場合（APPEND の例を参照）、これらのデフォルトモードで正しいインクリメンタルトランスフォームが実行されます。デフォルト以外のモードで読み取りや書き込みを行うには、transform() デコレーターを使用する必要があります。

トランスフォームが、SNAPSHOT トランザクションを含むデータセットを入力として受け取るものの、それがトランスフォームをインクリメンタルに実行できるかどうかに影響しない場合（参照テーブルなど）は、トランスフォームが全体の SNAPSHOT として実行されるのを避けるために、snapshot_inputs 引数を確認してください。

パラメーター

• require_incremental (ブール値 ↗、任意)

  • True の場合、トランスフォームは、過去に一度も実行されていない場合を除き、非インクリメンタルでの実行を拒否します。これは、すべての出力データセットにコミット済みトランザクションがないことに基づいて判断されます。

• semantic_version (int ↗、任意)

  • デフォルトは 1 です。この数値は、トランスフォームの意味的な性質を表します。既存の出力が無効になるような形でトランスフォームのロジックが変わる場合は、必ずこの値を変更する必要があります。この値を変更すると、次回のトランスフォームは非インクリメンタルで実行されます。

• snapshot_inputs (str のリスト、任意)

  • SNAPSHOT トランザクションが発生しても、トランスフォームの現在の出力が無効にならない入力です。たとえば、参照テーブルが更新されても、以前に計算した出力が誤っていることにはなりません。これら以外のすべての入力で、データの追加のみが行われたか、新しいデータがない場合、トランスフォームはインクリメンタルに実行されます。snapshot_inputs を読み取る場合、IncrementalTransformInput は入力データセットの現在のビューのみを公開します。

• allow_retention (ブール値、任意)

  • True の場合、foundry-retention による削除はインクリメンタル処理を妨げません。

• strict_append (ブール値、任意)

  • strict_append パラメーターが True に設定され、入力データセットがインクリメンタルである場合、基盤となる Foundry のトランザクションの種類は APPEND に設定され、インクリメンタル書き込みには APPEND トランザクションが使用されます。既存のファイルを上書きしようとすると、例外が発生します。 入力データセットがインクリメンタルでない場合、strict_append は SNAPSHOT として実行されます。コードが APPEND としてインクリメンタルに実行されることを保証するには、require_incremental=True を使用する必要があります。既存のファイルを上書きしようとすると、上書きに成功します。 書き込み操作では、Parquet の概要メタデータや Hadoop の SUCCESS ファイルのような補助ファイルも含め、いかなるファイルも上書きできません。すべての Foundry 形式のインクリメンタル書き込みは、strict_append モードをサポートしているはずです。

• v2_semantics (ブール値、任意)

  • デフォルトは False です。v2 のインクリメンタルセマンティクスを使用するには、True に設定します。V1 と v2 のセマンティクスは同じように動作するはずであり、すべてのユーザーに v2 を推奨します。メディアセットを含む、Catalog 以外のインクリメンタル入力と出力をサポートするのは、v2 セマンティクスのみです。詳細は、インクリメンタルメディアセットを参照してください。

発生する例外

• TypeError ↗
  • ラップされたオブジェクトが Transform オブジェクトでない場合。
• KeyError ↗
  • スナップショット入力が Transform オブジェクトに存在しない場合。

---

lightweight

transforms.api.lightweight(cpu_cores=2, memory_mb=None, memory_gb=16, gpu_type=None, container_image=None, container_tag=None, container_shell_command=None)

• デコレートされたトランスフォームを、コンテナトランスフォームのインフラストラクチャ上で実行するためのデコレーターです。
• lightweight デコレーターは、Transform をラップするために使用する必要があります。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
>>> @lightweight
... @transform(
...     my_input=Input('/input'),
...     my_output=Output('/output')
... )
... def compute_func(my_input, my_output):
...     my_output.write_pandas(my_input.pandas())

>>> @lightweight()
... @transform(
...    my_input=Input('/input'),
...    my_output=Output('/output')
... )
... def compute_func(my_input, my_output):
...     for file in my_input.filesystem().ls():
...         with my_input.filesystem().open(file.path) as f1:
...             with my_output.filesystem().open(file.path, "w") as f2:
...                 f2.write(f1.read())

>>> @lightweight(memory_gb=3.5)
... @transform_pandas(
...     Output('/output'),
...     my_input=Input('/input')
... )
... def compute_func(my_input):
...     return my_input

>>> @lightweight(container_image='my-image', container_tag='0.0.1')
... @transform(my_output=Output('ri...my_output'))
... def run_data_generator_executable(my_output):
...     os.system('$USER_WORKING_DIR/data_generator')
...     my_output.write_table(pd.read_csv('data.csv'))

軽量トランスフォームは、Spark を使用せず、単一ノード上で実行されるトランスフォームです。軽量トランスフォームは、小規模から中規模のデータセットに対して、より高速でコスト効率に優れています。ただし、軽量トランスフォームがサポートするのは、pandas やファイルシステム API など、通常のトランスフォームの API の一部のみです。一方で、データセットにアクセスするためのメソッドはより多く提供されます。軽量トランスフォームの詳細は、ドキュメントを確認してください。

container_image、container_tag、または container_shell_command のいずれかが設定されている場合、container_image と container_tag の両方を設定する必要があります。container_shell_command が設定されていない場合は、Python 環境をセットアップし、トランスフォームに指定されたユーザーコードを実行するデフォルトのエントリーポイントが使用されます。

container_* 引数を指定することを、bring-your-own-container（BYOC）ワークフローと呼びます。このワークフローでは、実行時にユーザーのコードリポジトリ内のすべてのファイルが $USER_WORKING_DIR/user_code 内で利用可能になり、Python 環境も利用可能になることが保証されます。

container_image および container_tag で指定するイメージは、コードリポジトリの Artifacts の取得元リポジトリから利用可能である必要があります。詳細は、BYOC のドキュメントを確認してください。

パラメーター

• cpu_cores (Float、任意)
  • トランスフォームに割り当てる CPU コア数です。小数も指定できます。デフォルトは 2 です。
• memory_mb (Float、任意)
  • コンテナに割り当てるメモリー量（MB 単位）です。memory_gb または memory_mb のいずれかを指定できますが、両方は指定できません。
• memory_gb (Float、任意)
  • コンテナに割り当てるメモリー量（GB 単位）です。デフォルトは 16GB です。
• gpu_type (str、任意_)
  • トランスフォームに割り当てる GPU の種類です。
• container_image (str、任意_)
  • トランスフォームのコンテナに使用するイメージです。
• container_tag (str、任意_)
  • トランスフォームのコンテナに使用するイメージタグです。
• container_shell_command (str、任意_)
  • コンテナ内で実行するシェルコマンドです。指定しない場合はデフォルト値が生成され、コンテナの起動後にユーザーのトランスフォームが実行されます。

---

transform

transforms.api.transform(ios)

• 計算関数を Transform オブジェクトとしてラップします。

• transform デコレーターは、計算関数から Transform オブジェクトを構築するために使用します。入力と出力に使用する名前は、ラップする計算関数のパラメーター名にする必要があります。計算時には、入力と出力がそれぞれ TransformInput オブジェクトと TransformOutput オブジェクトとして関数に渡されます。計算関数は、出力へのデータの書き込みを行う必要があります。

Copied!
1
2
3
4
5
6
7
8
9
10
>>> @transform(
...     first_input=Input('/path/to/first/input/dataset'),
...     second_input=Input('/path/to/second/input/dataset'),
...     first_output=Output('/path/to/first/output/dataset'),
...     second_output=Output('/path/to/second/output/dataset'),
... )
... def my_compute_function(first_input, second_input, first_output, second_output):
...     # type: (TransformInput, TransformInput, TransformOutput, TransformOutput) -> None
...     first_output.write_dataframe(first_input.dataframe())
...     second_output.write_dataframe(second_input.dataframe())

パラメーター

• ios (Input または Output)
  • 名前付きの Input と Output の仕様で構成される kwargs（キーワード引数）。

必要に応じて、計算関数内で TransformContext と対応する SparkSession にアクセスすることもできます。これを使用すると、空のデータフレームの作成や Spark 設定の適用などを行えます。可能な場合は、SparkSession オブジェクトを通じて Spark の設定値を設定するのではなく、既存のデフォルトの Spark プロファイルを使用することを推奨します。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
>>> @transform(
...     output=Output('/path/to/first/output/dataset'),
... )
... def my_compute_function(ctx, output):
...     # type: (TransformContext, TransformOutput) -> None
...
...     # この例では、Spark セッションを使って空のデータフレームを作成します。
...     columns = [
...         StructField("col_a", StringType(), True)
...     ]
...     empty_df = ctx.spark_session.createDataFrame([], schema=StructType(columns))
...
...     output.write_dataframe(empty_df)

---

transform_df

transforms.api.transform_df(output, inputs)

• ラップした計算関数をデータフレームトランスフォームとして登録します。
• transform_df デコレーターは、pyspark.sql.DataFrame ↗ オブジェクトを受け取り、返す計算関数から Transform オブジェクトを構築するために使用します。transform() デコレーターと同様に、入力名が計算関数のパラメーター名になります。ただし、transform_df は位置引数として単一の Output 仕様のみを受け取ります。計算関数の戻り値も DataFrame ↗ であり、単一の出力データセットに自動的に書き込まれます。

Copied!
1
2
3
4
5
6
7
8
>>> @transform_df(
...     Output('/path/to/output/dataset'),  # 名前のない Output 仕様
...     first_input=Input('/path/to/first/input/dataset'),
...     second_input=Input('/path/to/second/input/dataset'),
... )
... def my_compute_function(first_input, second_input):
...     # type: (pyspark.sql.DataFrame, pyspark.sql.DataFrame) -> pyspark.sql.DataFrame
...     return first_input.union(second_input)

パラメーター

• output (Output)
  • トランスフォームの単一の Output 仕様。
• inputs (Input)
  • 名前付きの Input 仕様で構成される kwargs（キーワード引数）。

必要に応じて、計算関数内で TransformContext と対応する SparkSession にアクセスすることもできます。これを使用すると、空のデータフレームの作成や Spark 設定の適用などを行えます。可能な場合は、SparkSession オブジェクトを通じて Spark の設定値を設定するのではなく、既存のデフォルトの Spark プロファイルを使用することを推奨します。

Copied!
1
2
3
4
5
6
7
8
9
10
11
>>> @transform_df(
...     Output('/path/to/output/dataset')
... )
... def my_compute_function(ctx):
...     # type: (TransformContext) -> pyspark.sql.DataFrame
...
...     # この例では、Spark セッションを使って空のデータフレームを作成します。
...     columns = [
...         StructField("col_a", StringType(), True)
...     ]
...     return ctx.spark_session.createDataFrame([], schema=StructType(columns))

---

transform_pandas

transforms.api.transform_pandas(output, inputs)

• ラップした計算関数を pandas トランスフォームとして登録します。

transform_pandas デコレーターは、pandas.DataFrame ↗ オブジェクトを受け取り、返す計算関数から Transform オブジェクトを構築するために使用します。このデコレーターは transform_df() デコレーターと似ていますが、計算前に pyspark.sql.DataFrame ↗ オブジェクトが pandas.DataFrame ↗ オブジェクトに変換され、計算後に元に戻されます。

Copied!
1
2
3
4
5
6
7
8
>>> @transform_pandas(
...     Output('/path/to/output/dataset'),  # 名前のない Output 仕様
...     first_input=Input('/path/to/first/input/dataset'),
...     second_input=Input('/path/to/second/input/dataset'),
... )
... def my_compute_function(first_input, second_input):
...     # type: (pandas.DataFrame, pandas.DataFrame) -> pandas.DataFrame
...     return first_input.concat(second_input)

transform_pandas デコレーターは、メモリーに収まるデータセットにのみ使用してください。より大きなデータセットを pandas に変換する前に絞り込みたい場合は、transform_df() デコレーターと pyspark.sql.SparkSession.createDataFrame() ↗ メソッドを使用してトランスフォームを記述してください。

Copied!
1
2
3
4
5
6
7
8
9
10
>>> @transform_df(
...     Output('/path/to/output/dataset'),  # 名前のない Output 仕様
...     first_input=Input('/path/to/first/input/dataset'),
...     second_input=Input('/path/to/second/input/dataset'),
... )
... def my_compute_function(ctx, first_input, second_input):
...     # type: (pyspark.sql.DataFrame, pyspark.sql.DataFrame) -> pyspark.sql.DataFrame
...     pd = first_input.filter(first_input.county == 'UK').toPandas()
...     # PySpark DataFrame に戻す前に、データの一部に対して pandas の操作を実行します。
...     return ctx.spark_session.createDataFrame(pd)

パラメーター

• output (Output)
  • トランスフォームの単一の Output 仕様。
• inputs (Input)
  • 名前付きの Input 仕様で構成される kwargs（キーワード引数）。

---

transform_polars

transforms.api.transform_polars(output, inputs)

• ラップした計算関数を Polars トランスフォームとして登録します。

transform_polars デコレーターは、polars.DataFrame ↗ オブジェクトを受け取り、返す計算関数から Transform オブジェクトを構築するために使用します。このデコレーターは transform_df() デコレーターと似ていますが、ユーザーコードには polars.DataFrame ↗ オブジェクトが渡されます。

transform_polars デコレーターは、@lightweight デコレーターの薄いラッパーにすぎません。これを使用すると軽量トランスフォームが作成されますが、通常のトランスフォームや軽量トランスフォームに一般的に備わっている機能のすべてに対応しているわけではありません。軽量トランスフォームの詳細については、軽量トランスフォームのドキュメントを参照してください。

たとえば、transform_polars トランスフォームに割り当てられるリソースを変更するには、まず lightweight トランスフォームに切り替える必要があります。

次のような transform_polars トランスフォームを使用する代わりに、

@transform_polars(
    output=Output('/Project/folder/output'),
    dataset=Input('/Project/folder/input'),
)
def compute(output, dataset):
    # ここでは、dataset はすでに polars のデータフレームです。

lightweight トランスフォームを使用してください。

@lightweight(cpu_cores=4, memory_gb=16)
@transform(
    output=Output('/Project/folder/output'),
    dataset=Input('/Project/folder/input'),
)
def compute(output, dataset):
    my_df = dataset.polars(lazy=True)

Spark プロファイルとトランスフォームのその他の一部の機能は、@lightweight トランスフォームでは使用できないため、@transform_polars でも使用できません。

Copied!
1
2
3
4
5
6
7
8
>>> @transform_polars(
...     Output('ri.main.foundry.dataset.out'),  # 名前のない Output 仕様
...     first_input=Input('ri.main.foundry.dataset.in1'),
...     second_input=Input('ri.main.foundry.dataset.in2'),
... )
... def my_compute_function(ctx, first_input, second_input):
...     # type: (polars.DataFrame, polars.DataFrame) -> polars.DataFrame
...     return first_input.join(second_input, on='id', how="inner")

パラメーター

• output (Output)
  • トランスフォームの単一の Output 仕様。
• inputs (Input)
  • 名前付きの Input 仕様で構成される kwargs（キーワード引数）。