軽量トランスフォーム API の進化

軽量コンピュートのトランスフォーム API は、より簡潔な構文を使えるように進化してきました。

軽量 API の進化の詳細については、このページを参照してください。

軽量トランスフォームの記述をはじめるには、はじめに、または Python トランスフォームの Polars と pandas の例を参照してください。これらのページでは、軽量トランスフォームをデフォルトのコンピュートオプションとして使用する方法を説明しています。

レガシー構文：@lightweight デコレーター

軽量コンピュートの元の構文は、@lightweight デコレーターです。この構文は、引き続き完全にサポートされています。

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


@lightweight
@transform(
    output=Output("/path/data/output"),
    input=Input("/path/data/input"),
)
def clean(output, input):
    df = input.pandas()
    output.write_table(df)

Lightweight の更新された構文

軽量トランスフォームを使用するための新しい推奨構文は、@transform.using です。この API では、軽量トランスフォーム用の追加のインポートが不要になり、デフォルトで軽量トランスフォームを作成する手順が簡素化されます。

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

transforms のバージョンについては、transforms のバージョンの概要で詳しくはこちらをご覧ください。

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


@transform.using(
    output=Output("/path/data/output"),
    input=Input("/path/data/input"),
)
def clean(output, input):
    df = input.pandas()
    output.write_table(df)

まとめると、以下のいずれの構文でも Lightweight トランスフォームを作成できます。

• 更新された API:
  @transform.using(...)
• Lightweight を明示的に指定する更新された API:
  @transform.lightweight(...)
• @lightweight デコレーターを使用するレガシー API:
  @lightweight
@transform(...)

Spark の更新された構文

新しいデフォルトの軽量 API に加え、Spark 用の新しいトランスフォーム API もあります。推奨構文は @transform.spark.using です。

新しい API は、transforms バージョン 3.95.0 以降で利用可能です。Code Repositories で利用できるようにするには、リポジトリのアップグレードガイドに従ってリポジトリをアップグレードします。transformsLangPythonPluginVersion が 1.1003.0 以上であることを確認してください。

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


@transform.spark.using(
    output=Output("/path/data/output"),
    input=Input("/path/data/input"),
)
def clean(output, input):
    df = input.dataframe()
    output.write_dataframe(df)

Databricks Connect の更新された構文

Databricks Connect 経由でコンピュート処理を Databricks にプッシュダウンするトランスフォームを定義するには、統合されたデコレーター @transform.databricks.using を使用します。この標準化された構文は、以前の @databricks と @transform デコレーターを組み合わせたパターンに代わるものです。

設定メソッドを連結して、コンピュートと依存関係を指定できます。

• with_compute(cluster_id="<cluster-id>")：特定の Databricks コンピュートクラスターに接続します
• with_dependencies(dependencies=["<dependency>"])：Databricks コンピュート上で UDF に必要な Python パッケージをインストールします

デフォルトでは、Databricks Connect はサーバーレスコンピュートを使用してセッションを確立します。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from pyspark.sql.functions import col
from transforms.api import transform
from transforms.tables import DatabricksInput, DatabricksOutput, TableInput, TableOutput

@transform.databricks.using(
    source_table=TableInput('/Project/folder/input'),
    output_table=TableOutput(
        '/Project/folder/output',           # Foundry に仮想テーブルを登録します
        'ri.magritte..source.1234',         # Databricks ソース接続の RID
        'CATALOG.SCHEMA.TABLE',             # Databricks 内の対象の Unity Catalog テーブル
    ),
).with_compute(cluster_id="<cluster-id>")
def compute(source_table: DatabricksInput, output_table: DatabricksOutput):
    df = source_table.dataframe()
    df = df.filter(col('id') > 1)
    output_table.write_dataframe(df)

追加の Python 依存関係を必要とする UDF では、with_dependencies を使用して依存関係を設定します。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from pyspark.sql.functions import udf
from pyspark.sql.types import StringType
from transforms.api import transform
from transforms.tables import DatabricksOutput, TableOutput

@udf(returnType=StringType())
def get_joke():
    from pyjokes import get_joke
    return get_joke()

@transform.databricks.using(
    output_table=TableOutput(
        '/Project/folder/output',
        'ri.magritte..source.1234',
        'CATALOG.SCHEMA.TABLE',
    ),
).with_dependencies(dependencies=["pyjokes<1"])
def compute(output_table: DatabricksOutput):
    df = output_table.spark_session.range(1, 10)
    df = df.withColumn('jokes', get_joke())
    output_table.write_dataframe(df)

@transform.using のバージョンエラーをトラブルシューティングする

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

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

この問題を解決するには、次の方法を試してください。

• リポジトリのアップグレードガイドに従って、リポジトリの transforms ライブラリとリポジトリテンプレートを最新バージョンにアップグレードします。
• レガシー構文に切り替えます。