好みの統合開発環境 (IDE) でPythonパイプライン ソース コードを作成し、テスト用にローカルで実行し、ローカル環境を離れることなく、Azure Databricks ワークスペースで更新プログラムを検証、デプロイ、および実行できます。
Lakeflow パイプラインは、Apache Spark™ 宣言型パイプラインのスーパーセットです。 Apache Spark 宣言型パイプライン API のみを使用するコードはローカルとAzure Databricksの両方で実行されますが、AUTO CDCや期待など、Lakeflow パイプライン固有の機能を使用するコードは、Azure Databricksでのみ実行されます。 機能の違いについては、Lakeflow パイプラインPython言語リファレンスを参照してください。
Azure Databricks ワークスペースでの対話型の開発とテストには、Lakeflow Pipelines エディターを使用します。 Lakeflow パイプライン エディターを使用した ETL パイプラインの開発とデバッグを参照してください。
IDE サポートを使用してパイプライン コードを記述する
pyspark.pipelinesとしてインポートされたdp モジュールを使用してパイプライン コードを記述します。
from pyspark import pipelines as dp
モジュールは Apache Spark の一部であるため、IDE では、記述時に構文チェック、オートコンプリート、および型チェックが提供されます。 Apache Spark 宣言型パイプライン コードは、通常、Azure Databricksに変更を加えずに実行されます。 Lakeflow パイプラインで同じインポート コマンドを実行すると、Azure Databricksバージョンのpipelinesがインポートされます。 完全な Lakeflow パイプラインPythonリファレンスについては、Lakeflow パイプラインPython言語リファレンスを参照してください。
局所検査のための別個変換ロジック
パイプラインコードをローカルでテスト可能にする最も効果的な方法は、変換ロジックを単なるPySpark関数にまとめ、 dp デコレーターとは別にすることです。 DataFrameを受け取り返す関数はLakeflowパイプラインのランタイムに依存しないので、ローカルマシンでユニットテストを行えば、他のApache Sparkコードと同様に pytest でテストできます。 デコレータを適用した関数は最小限に保ち、ロジックをインポートしてそれを呼び出すだけにします:
# transformations/clean.py — pure PySpark, unit-testable on its own
def clean_orders(df):
return df.filter("quantity > 0").withColumn("amount_usd", df.amount.cast("double"))
# pipeline file — a thin dp wrapper that imports and calls the logic
from pyspark import pipelines as dp
from transformations.clean import clean_orders
@dp.table(name="orders_silver")
def orders_silver():
return clean_orders(spark.readStream.table("orders_bronze"))
共有ロジックをホイールとしてパッケージ化し、パイプライン間で再利用することができます。 ユニットテストの執筆と実行の詳細な手順については、「 パイプラインのユニットテスト」をご覧ください。
テスト用にパイプラインをローカルで実行する
また、Azure Databricksで実行する前に、パイプラインをローカルで実行してコードを開発およびテストすることもできます。
spark-pipelines コマンド ライン インターフェイスを使用して、ローカル Apache Spark を使用してパイプラインを初期化、検証、および実行します。 Apache Spark ドキュメントの Spark 宣言型パイプライン プログラミング ガイド を参照してください。
完全なパイプラインは3つの補完的なテスト層を使用し、そのうち2つはローカルで実施可能です。
- 変換ロジックのユニットテストを行い、上記のPySpark関数に対して
pytestを使います。 これらはパイプラインの実行時間を必要としません。 パイプラインの単体テストを参照してください。 - パイプライングラフ、ソースコード、データセット参照の検証（ドライラン）を行います。
databricks pipelines dry-runをローカルで、またはspark-pipelinesを使用してワークスペースに対して実行でき、データを書き込むことはありません。 - 期待値(Expectations)は、すべての実行行でデータ品質ルールを評価するものです。 これらはLakeflowパイプラインのランタイム機能であるため、ローカルでは動作せず、Azure Databricks上でのみ動作します。 「パイプラインの期待値を使用してデータ品質を管理する」を参照してください。
Lakeflow パイプラインに固有の機能をローカルで実行またはテストすることはできません。 これには、期待値と AUTO CDC 関数が含まれます。
ローカル環境からAzure Databricksでパイプラインを実行する
databricks pipelines コマンド グループを使用して、ターミナルから直接、ワークスペースでパイプラインの更新を検証、デプロイ、および実行します。
databricks pipelines init # scaffold a pipeline project
databricks pipelines dry-run # validate the pipeline graph without publishing data
databricks pipelines deploy # deploy the project to your workspace
databricks pipelines run # run an update
パイプラインの更新は、パイプライン用に構成されたコンピューティングを使用して、ローカル コンピューターではなく、Azure Databricks ワークスペースで実行されます。 これらのコマンドは、宣言型オートメーション バンドル bundle コマンドと相互運用できるため、単純なプロジェクトから始めて、バンドル構成と CI/CD プラクティスを採用することができます。 CLI をインストールして構成するには、 Databricks CLI のインストールまたは更新に関する記事を参照してください。 完全なコマンド リファレンスについては、コマンド グループpipelines参照してください。 詳細なチュートリアルについては、「 宣言型オートメーション バンドルを使用したパイプラインの開発」を参照してください。
IDE からワークスペースにパイプライン コードを同期する
次の表は、ローカル IDE と Azure Databricks ワークスペースの間でパイプライン ソース コードを同期するためのオプションをまとめたものです。
| ツールまたはパターン | 詳細 |
|---|---|
Databricks CLI (pipelines コマンド グループ) |
databricks pipelines コマンドを使用して、ローカル環境からパイプライン プロジェクトをデプロイして実行します。
コマンド グループpipelines参照してください。 |
| 宣言型オートメーション バンドル | 宣言型オートメーション バンドルを使用して、1 つのソース コード ファイルから複数のパイプライン、ジョブ、ソース コード ファイルの構成まで、複雑なパイプライン資産をデプロイします。 「パイプラインをバンドル プロジェクトに変換する」を参照してください。 |
| Databricks IDE拡張 | Azure Databricksは、ローカル IDE ファイルとワークスペース ファイル間の簡単な同期を含むVisual Studio Codeとの統合を提供します。 この拡張機能には、宣言型オートメーション バンドルを使用してパイプライン資産をデプロイするためのツールも用意されています。 DatabricksのIDE拡張を参照してください。 |
| ワークスペース ファイル | Databricks ワークスペース ファイルを使用して、パイプラインのソース コードを Databricks ワークスペースにアップロードし、そのコードをパイプラインにインポートできます。 「ワークスペース ファイルとは」を参照してください。 |
| Git フォルダー | Git フォルダーを使用すると、ローカル環境と Azure Databricks ワークスペースの間で、Git リポジトリを仲介者として使用してコードを同期できます。 Azure Databricks Git フォルダーを参照してください。 |