Lakeflow パイプライン Python インターフェイスは、pyspark.pipelines モジュールで定義され、dpとしてインポートされます。
このモジュールは、データセット、フロー、シンク、データ品質の期待値を定義するためのデコレーターや関数を提供します。 サーバーレスパイプライン では、クラスタを構成しなくてもこれらの定義を実行できます。
- 概念情報とパイプラインに Python を使用する方法の概要については、「Python を使用した パイプライン コードの開発」を参照してください。
- SQL リファレンスについては、 パイプライン SQL 言語リファレンスを参照してください。
- 自動ローダーの構成に固有の詳細については、「 自動ローダーとは」を参照してください。
pipelines モジュールの概要
関数Python Lakeflow パイプラインは、pyspark.pipelines モジュールで定義されます (dpとしてインポートされます)。 Python API で実装されたパイプラインでは、次のモジュールをインポートする必要があります。
from pyspark import pipelines as dp
注
パイプラインモジュールの大部分はパイプラインの文脈でのみ利用可能です。 パイプライン コードの編集の詳細については、「 Lakeflow Pipelines Editor を使用した ETL パイプラインの開発とデバッグ」を参照してください。
また、パイプラインの外側、ノートブックのサーバーレス汎用計算で、単独のマテリアルビューやストリーミングテーブルを定義するために @materialized_view や @table を適用することもできます。 「 パイプラインデコレーター付きのテーブルを定義する」を参照してください。
Apache Spark™ パイプライン
Apache Spark には、 モジュールから入手できる Spark 4.1 以降のpyspark.pipelinesが含まれています。 Databricks ランタイムは、マネージド 運用で使用するための追加の API と統合によって、これらのオープン ソース機能を拡張します。
オープンソースの pipelines モジュールで記述されたコードは、Azure Databricks で変更なしで実行されます。 次の機能は Apache Spark の一部ではありません。
dp.create_auto_cdc_flowdp.create_auto_cdc_from_snapshot_flow@dp.expect(...)
pipelines モジュールは、以前は Azure Databricks でdltと呼ばれていました。 Apache Spark との違いの詳細と詳細については、「@dltの変更点」を参照してください。
データセット定義の関数
パイプラインでは、具体化されたビューやストリーミング テーブルなどのデータセットを定義するために Python デコレーターを使用します。 データセットを定義する関数を参照してください。
API リファレンス
- append_flow
- create_auto_cdc_flow
- create_auto_cdc_from_snapshot_flow
- create_table
- create_sink
- ストリーミングテーブルを作成
- 期待
- foreach_batch_sink
- foreach_sink
- マテリアライズドビュー
- replace_flow
- table
- temporary_view
- update_flow
Python パイプラインのコーディング要件
Pythonインターフェースでパイプラインを実装する際の重要な要件は以下の通りです:
- パイプラインは計画時と実行時にソースコードを複数回評価します。 データセットを定義する Python 関数には、テーブルまたはビューを定義するために必要なコードのみを含める必要があります。 データセット定義に含まれる任意の Python ロジックは、予期しない動作につながる可能性があります。
- データセット定義にカスタム監視ロジックを実装しないでください。 イベント フックを使用したパイプラインのカスタム監視の定義を参照してください。
- データセットの定義に使用する関数は、Spark DataFrame を返す必要があります。 返された DataFrame に関連しないロジックをデータセット定義に含めないでください。
- パイプライン データセット コードの一部として、ファイルまたはテーブルを保存または書き込むメソッドは使用しないでください。
パイプライン コードで使用してはならない Apache Spark 操作の例を次に示します。
collect()count()toPandas()save()saveAsTable()start()toTable()
@dltはどうなりましたか?
以前は、Azure Databricks はパイプライン機能をサポートするために dlt モジュールを使用しました。
dlt モジュールは、pyspark.pipelines モジュールに置き換えられました。
dltは引き続き使用できますが、Databricks ではpipelinesを使用することをお勧めします。
DLT、Lakeflowパイプライン、Apache Spark™宣言型パイプラインの違い
以下の表は、DLT、Lakeflowパイプライン、Apache Spark™宣言型パイプライン間の構文と機能の違いを示しています。
LakeflowパイプラインとApache Spark™ Declarative Pipelinesとの共通機能と、Lakeflowパイプラインで追加される機能の比較については、Apache Spark Declarative Pipelinesをご覧ください。
パイプライン構成と SDP プロジェクト仕様のプロパティごとのマッピングについては、「 パイプライン のプロパティ リファレンス」を参照してください。
注
Databricks のドキュメントでは、Databricks の製品は Lakeflow パイプラインと呼ばれており、その基盤となっている Apache Spark プロジェクトは Apache Spark™ Declarative Pipelines (SDP) です。 両者は相互運用可能ですが、特徴は異なります。 例えば、 AUTO CDC APIはLakeflowパイプラインでのみ利用可能です。
| Area | DLT 構文 | SDP 構文 (該当する場合は Lakeflow と Apache) | Apache Spark で使用可能 |
|---|---|---|---|
| インポート | import dlt |
from pyspark import pipelines (as dp、必要に応じて) |
イエス |
| ストリーミング テーブル |
@dlt.table ストリーミング データフレームを使用する |
@dp.table |
イエス |
| マテリアライズド・ビュー |
@dlt.table バッチ データフレームを使用する |
@dp.materialized_view |
イエス |
| 表示 | @dlt.view |
@dp.temporary_view |
イエス |
| 追加フロー | @dlt.append_flow |
@dp.append_flow |
イエス |
| 更新フロー | 利用不可 | @dp.update_flow |
いいえ |
| SQL – ストリーミング | CREATE STREAMING TABLE ... |
CREATE STREAMING TABLE ... |
イエス |
| SQL – マテリアライズド | CREATE MATERIALIZED VIEW ... |
CREATE MATERIALIZED VIEW ... |
イエス |
| SQL – フロー | CREATE FLOW ... |
CREATE FLOW ... |
イエス |
| イベント ログ | spark.read.table("event_log") |
spark.read.table("event_log") |
いいえ |
| 変更の適用 (CDC) | dlt.apply_changes(...) |
dp.create_auto_cdc_flow(...) |
いいえ |
| Expectations | @dlt.expect(...) |
dp.expect(...) |
いいえ |
| 連続モード | 継続的トリガーを使用したパイプライン構成 | 継続的トリガーを使用したパイプライン構成 | いいえ |
| シンク | @dlt.create_sink(...) |
dp.create_sink(...) |
イエス |
ForEachBatch シンク |
利用不可 | @dp.foreach_batch_sink(...) |
いいえ |
ForEach シンク |
利用不可 | @dp.foreach_sink(...) |
いいえ |