Databricks へのコンピュートプッシュダウン

Foundry では、仮想テーブルを使用する際に、コンピュートを Databricks にプッシュダウンできます。同じソースに登録された Databricks 仮想テーブルをパイプラインの入力と出力として使用する場合、コンピュートを完全に Databricks に委譲できます。

このドキュメントでは、Databricks 環境内で完全に実行できる Python トランスフォームを Code Repositories で開発する手順を説明します。この機能は Databricks Connect を利用します。Databricks Connect の機能と制限の詳細については、Databricks の公式ドキュメント ↗を参照してください。

クイックスタート

この例では、Python トランスフォームパイプラインで Databricks トランスフォームを使用する方法を示します。@transform を介して PySpark を使用する、次の Spark パイプラインがあるとします。

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


@transform.spark.using(
    source_table=Input('/Project/folder/input'),
    output_table=Output('/Project/folder/output'),
)
def compute(source_table: TransformInput, output_table: TransformOutput):
    df = source_table.dataframe()
    df = df.filter(col('id') > 1)
    output_table.write_dataframe(df)

これを Databricks トランスフォームに変更するには、次の手順を実行する必要があります。

Python トランスフォームでテーブルを使用するための前提条件を確認します。
ライブラリタブから databricks-connect をインストールします。
@databricks デコレーターをインポートし、トランスフォームに適用します。

詳細については、以下のドキュメントのセットアップセクションを参照してください。

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from pyspark.sql.functions import col
from transforms.api import transform
from transforms.tables import (
    databricks,
    DatabricksInput,
    DatabricksOutput,
    TableInput,
    TableOutput,
)


@transform.databricks.using(
    source_table=TableInput('/Project/folder/input'),
    output_table=TableOutput(
        '/Project/folder/output', # Databricks の出力を Foundry の仮想テーブルとして登録する場所
        "ri.magritte..source.1234", # Databricks のソース接続
       "CATALOG.SCHEMA.TABLE", # Databricks でトランスフォームの出力が書き込まれる場所
    ),
)
def compute(source_table: DatabricksInput, output_table: DatabricksOutput):
    df = source_table.dataframe()
    df = df.filter(col('id') > 1)
    output_table.write_dataframe(df)

Databricks Connect は pyspark.sql.DataFrame ↗ API を使用するため、コード自体のロジックはほとんど変わりません。主な違いは、トランスフォームが Databricks で稼働する Spark コンピュートを使用することです。

トランスフォームの入力と出力となる Databricks 仮想テーブルを参照するために、Input と Output の代わりに TableInput と TableOutput パラメーターを使用します。トランスフォーム関数には、Databricks のテーブルの読み取りと書き込みに使用できる DatabricksInput と DatabricksOutput パラメーターが渡されます。

上記の @databricks デコレーターは、TableInput と TableOutput パラメーターにのみ対応しています。トランスフォームの入力と出力として参照されるテーブルは、同じ Databricks ソースに登録されている必要があります。

Databricks へのコンピュートプッシュダウンを使用する場合、@incremental デコレーターによるインクリメンタル（差分処理）計算は現在サポートされていません。

セットアップ

Databricks へのコンピュートプッシュダウンを使用するには、次の手順を実行します。

Python コードリポジトリを作成します。
Python トランスフォームでテーブルを使用するための前提条件を確認します。
databricks-connect ライブラリをインストールします。サーバーレスコンピュートに完全に対応しているバージョン 16.4 の使用を推奨します。クラシックコンピュートを使用する場合は、Databricks クラスターのバージョンに対応するバージョンを使用する必要があります。詳細については、Databricks の公式ドキュメント ↗を参照してください。

databricks-connect ライブラリは、コードリポジトリ環境のライブラリタブからインストールできます。このライブラリは PyPI 経由でのみ利用可能で、Conda 経由では利用できません。または、conda_recipe/meta.yaml ファイルの pip requirements ブロックにライブラリを手動で追加できます。たとえば、databricks-connect のバージョン 16.4 をインストールするには、次を追加します。

Copied!1
2
3
requirements:
  pip:
   - databricks-connect >=16.4.0,<17.0.0

API の主な特徴

以下のセクションでは、Databricks トランスフォーム API と通常の Python トランスフォームの違いを説明します。@databricks を使用する前に、上記のセットアップ手順を実施してください。

Databricks トランスフォームでは、transforms.tables.TableInput および transforms.tables.TableOutput パラメーターを使用して、入力および出力の Databricks 仮想テーブルを参照します。TableInput は、ファイルパスまたは RID で仮想テーブルを参照できます。TableOutput には、以下が必要です。

仮想テーブルを作成する Foundry 内のファイルパス。
仮想テーブルを登録するソース。
テーブルを書き込む Databricks 内の場所。これは、DatabricksTable("<catalog>", "<schema>", "<table>") または "<catalog>.<schema>.<table>" の構文で指定できます。catalog、schema、table は、Unity Catalog のテーブル識別子の3階層の名前空間構造に対応します。詳細については、Databricks の公式ドキュメント ↗を参照してください。

DatabricksInput および DatabricksOutput の API は、通常の Python トランスフォームと同様です。.dataframe() は、入力テーブルを pyspark.sql.DataFrame として読み込みます。.write_dataframe() は、pyspark.sql.DataFrame を出力テーブルに書き込みます。トランスフォームのロジック自体は、PySpark の変換を使用して記述できます。

Databricks Connect では、PySpark のすべての機能がサポートされているわけではありません。機能の利用可否と制限の詳細については、Databricks の公式ドキュメント ↗を参照してください。

出力テーブルの種類

以下のいずれかのテーブルの種類を使用して、Databricks に出力テーブルを書き込めます。

マネージド Delta： Databricks が管理するストレージに Delta Lake 形式で保存されるテーブル。
外部 Delta： 外部ストレージの場所に Delta Lake 形式で保存されるテーブル。
マネージド Iceberg： Databricks が管理するストレージに Iceberg 形式で保存されるテーブル。

DatabricksTable または write_dataframe の format および location パラメーターを使用して、出力テーブルの種類を設定できます。format と location を指定しない場合、テーブルはマネージド Delta テーブル（Databricks のデフォルトのテーブルの種類）として書き込まれます。出力をマネージド Iceberg テーブルとして書き込むには、format="iceberg" を使用します。出力を外部 Delta テーブルとして書き込むには、location="<STORAGE_LOCATION>" を使用します。ここで、<STORAGE_LOCATION> は、テーブルを保存するクラウドオブジェクトストレージ内の外部ストレージパスを指します。

以下の例は、異なる種類の3個の出力テーブルに書き込むトランスフォームを示しています。

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
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
from transforms.tables import (
    databricks,
    DatabricksTable,
    DatabricksInput,
    DatabricksOutput,
    TableInput,
    TableOutput,
)

@transform.databricks.using(
    source_table=TableInput('/Project/folder/input'),
    managed_delta_output_table=TableOutput(
        '/Project/folder/output',
        "ri.magritte..source.1234",
       "CATALOG.SCHEMA.MANAGED_DELTA_TABLE",
    ),
    managed_iceberg_output_table=TableOutput(
        '/Project/folder/output',
        "ri.magritte..source.1234",
       "CATALOG.SCHEMA.MANAGED_ICEBERG_TABLE",
    ),
    external_delta_output_table=TableOutput(
        '/Project/folder/output',
        "ri.magritte..source.1234",
        DatabricksTable(
            catalog="CATALOG",
            schema="SCHEMA",
            table="EXTERNAL_DELTA_TABLE",
            location="s3://some-bucket/path/to/table"
        ),
    ),
)
def compute(
    source_table: DatabricksInput,
    managed_delta_output_table: DatabricksOutput,
    managed_iceberg_output_table: DatabricksOutput,
    external_delta_output_table: DatabricksOutput,
):
    df = source_table.dataframe()

    # マネージド Delta テーブルに書き込みます
    managed_delta_output_table.write_dataframe(df)

    # format オプションをインラインで指定して、マネージド Iceberg テーブルに書き込みます
    managed_iceberg_output_table.write_dataframe(df, format="iceberg")

    # DatabricksTable で場所を指定した外部テーブルに書き込みます
    external_delta_output_table.write_dataframe(df)

テーブルの種類の詳細については、Databricks の公式ドキュメント ↗を参照してください。

コンピュート設定

デフォルトでは、サーバーレスコンピュートを使用して Databricks Connect セッションが確立されます。これは、@databricks(serverless=True) を使用することと同等です。

代わりに、with_compute(cluster_id="<cluster-id>") を使用して、特定のコンピュートクラスターへの接続を設定できます。

Databricks Connect のコンピュート設定については、Databricks の公式ドキュメント ↗を参照してください。

ユーザー定義関数（UDF）

Databricks Connect では、ユーザー定義関数（UDF）がサポートされています。UDF を含む PySpark DataFrame 操作を実行すると、Databricks Connect は UDF をシリアル化し、リクエストの一部としてサーバーに送信します。UDF の機能と制限の詳細については、Databricks の公式ドキュメント ↗を参照してください。

以下の例では、列の値を2乗する単純な UDF を定義しています。

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
from pyspark.sql.functions import col, udf
from pyspark.sql.types import IntegerType
from transforms.api import transform
from transforms.tables import (
    databricks,
    DatabricksInput,
    DatabricksOutput,
    TableInput,
    TableOutput,
)


@udf(returnType=IntegerType())
def double(x):
    return x * x


@transform.databricks.using(
    source_table=TableInput('/Project/folder/input'),
    output_table=TableOutput(
        '/Project/folder/output',
        "ri.magritte..source.1234",
       "CATALOG.SCHEMA.TABLE",
    ),
)
def compute(source_table: DatabricksInput, output_table: DatabricksOutput):
    df = source_table.dataframe()
    df = df.withColumn("doubled", double(col("id")))
    output_table.write_dataframe(df)

UDF の Python 依存関係を指定する

Databricks Connect では、UDF に必要な Python 依存関係の指定がサポートされています。これらの依存関係は、UDF の Python 環境の一部として Databricks コンピュートにインストールされます。with_dependencies(dependencies=["<dependency>"]) を使用して、UDF 環境にインストールする追加の依存関係を設定できます。

以下の例では、pyjokes パッケージに依存する UDF を定義しています。

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from pyspark.sql.functions import col, udf
from pyspark.sql.types import StringType
from transforms.api import transform
from transforms.tables import databricks, 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)

ModuleNotFoundError: No module named 'myproject' のようなエラーが発生した場合、通常は、UDF がローカル環境にのみ存在し、UDF を実行する Databricks Spark ワーカーでは利用できないコードを参照していることを意味します。このエラーを回避するには、UDF がローカルモジュールで定義された関数やクラスに直接依存しないようにしてください。以下のいずれかの方法で解決できます。

UDF の定義内に直接、必要なロジックをインライン化します。
@transform 関数の本体内で、UDF とその補助コードの両方を定義します。
UDF とその補助コードを、UDF への参照を返す関数でラップし、トランスフォーム内でこの関数を呼び出して UDF を作成します。

←

PREVIOUSBigQuery の計算プッシュダウン

NEXTSnowflake の計算プッシュダウン

→