Delta Live Tables 向けのパイプラインを開発する
パイプラインのコードの開発とテストは、他の Apache Spark ワークロードとは異なります。 この記事では、パイプラインのコードを開発するときのサポートされる機能、ベスト プラクティス、考慮事項の概要について説明します。 その他の推奨事項とベスト プラクティスについては、「ソフトウェア開発および DevOps のベスト プラクティスを Delta Live Table パイプラインに適用」を参照してください。
注
コードを検証したり、更新を実行したりするには、ソース コードをパイプライン構成に追加する必要があります。 「Delta Live Tables パイプラインの構成方法についてご覧ください。」
パイプラインのソース コードに対して有効なファイル
Delta Live Tables パイプラインのコードには、Python または SQL を使用できます。 Python と SQL 両方のソース コード ファイルを使って 1 つのパイプラインをサポートできますが、各ファイルに含めることができる言語は 1 つだけです。 「Python を使用してパイプライン コードを開発する」と「SQL を使用してパイプライン コードを開発する」をご覧ください。
ノートブックとワークスペース ファイルを使って、パイプラインのソース コードを指定できます。 ワークスペース ファイルは、任意の IDE または Databricks ファイル エディターで作成された Python または SQL スクリプトを表します。 「ワークスペース ファイルとは」を参照してください。
Python コードをモジュールまたはライブラリとして開発する場合は、コードをインストールしてインポートしてから、ソース コードとして構成された Python ノートブックまたはワークスペース ファイルから、メソッドを呼び出す必要があります。 「Delta Live Tables パイプラインの Python 依存関係を管理する」を参照してください。
注
Python ノートブックで任意の SQL コマンドを使う必要がある場合は、spark.sql("<QUERY>") という構文パターンを使って、SQL を Python コードとして実行できます。
Unity Catalog 関数を使うと、任意の Python ユーザー定義関数を登録して、SQL で使用できます。 「Unity Catalog のユーザー定義関数 (UDF)」をご覧ください。
Delta Live Tables の開発機能の概要
Delta Live Tables は、Azure Databricks の多くの機能を拡張して活用し、新しい機能と概念を導入するものです。 次の表は、パイプライン コードの開発をサポートする概念と機能の概要です。
| 機能 | 説明 |
|---|---|
| 開発モード | 新しいパイプラインは、既定で開発モードで実行するように構成されます。 Databricks では、対話型の開発とテストには開発モードを使うことをお勧めします。 「開発モードと実稼働モード」をご覧ください。 |
| 検証 | Validate の更新では、テーブルに対して更新を実行することなく、パイプライン ソース コードの正確性が検証されます。 「テーブルの更新を待たずにパイプラインにエラーがないかどうかを確認する」をご覧ください。 |
| ノートブック | Delta Live Tables パイプラインのソース コードとして構成されたノートブックでは、コードの検証と更新の実行のための対話型オプションが提供されます。 「ノートブックでの Delta Live Tables パイプラインの開発とデバッグ」をご覧ください。 |
| パラメーター | ソース コードとパイプライン構成でパラメーターを使って、テストと拡張性を簡素化します。 「Delta Live Tables パイプラインでパラメーターを使用する」をご覧ください。 |
| Databricks アセット バンドル | Databricks アセット バンドルを使うと、パイプラインの構成とソース コードをワークスペース間で移動できます。 「Delta Live Tables パイプラインを Databricks アセット バンドル プロジェクトに変換する」をご覧ください。 |
開発とテスト用のサンプル データセットを作成する
Databricks では、開発とテスト用のデータセットを作成し、予想されるデータと、形式の誤りや破損が含まれる可能性のあるレコードを使って、パイプラインのロジックをテストすることをお勧めします。 開発とテストに役立つデータセットを作成するには、次のような複数の方法があります。
- 運用データセットからデータのサブセットを選択する。
- PII を含むソースには、匿名化されたデータや人為的に生成されたデータを使用する。
- ダウンストリーム変換ロジックに基づいて、適切に定義された結果を含むテスト データを作成する。
- データ スキーマの想定から外れるレコードを作成し、潜在的なデータの破損、形式に誤りがあるレコード、アップストリーム データの変更を予測する。
たとえば、次のコードを使用してデータセットを定義するノートブックがあるとします。
CREATE OR REFRESH STREAMING TABLE input_data AS SELECT * FROM read_files("/production/data", "json")
次のようなクエリを使用して、特定のレコードを含むサンプル データセットを作成できます。
CREATE OR REFRESH MATERIALIZED VIEW input_data AS
SELECT "2021/09/04" AS date, 22.4 as sensor_reading UNION ALL
SELECT "2021/09/05" AS date, 21.5 as sensor_reading
次に示すのは、公開されたデータをフィルター処理して、開発またはテストのために運用データのサブセットを作成する方法の例です。
CREATE OR REFRESH MATERIALIZED VIEW input_data AS SELECT * FROM prod.input_data WHERE date > current_date() - INTERVAL 1 DAY
こうしたさまざまなデータセットを使用するには、変換ロジックを実装するノートブックを使用して複数のパイプラインを作成します。 各パイプラインは input_data データセットからデータを読み取ることができますが、環境に固有のデータセットを作成するノートブックを含むように構成されています。