Lakeflow宣言型パイプラインの構成

この記事では、ワークスペース UI を使用した Lakeflow 宣言型パイプラインの基本構成について説明します。

Databricksでは、サーバレスを使用して新しいパイプラインを開発することをお勧めします。 サーバレス パイプラインの設定手順については、 サーバレス パイプラインの設定を参照してください。

この記事の構成手順では、Unity Catalog を使用します。従来のHive metastoreでパイプラインを設定する手順については 、「従来のHive metastoreで Lakeflow宣言型パイプラインを使用する 」を参照してください。

この記事では、パイプラインの現在のデフォルトの公開モードの機能について説明します。2025 年 2 月 5 日より前に作成されたパイプラインでは、従来の公開モードとLIVE仮想スキーマが使用される可能性があります。LIVE スキーマ (レガシー)を参照してください。

注記

UI には、JSON で設定を表示および編集するオプションがあります。ほとんどの設定は、UI または JSON 仕様を使用して構成できます。一部の詳細オプションは、JSON 構成でのみ使用できます。

JSON 構成ファイルは、パイプラインを新しい環境にデプロイする場合や、CLI またはREST APIを使用する場合にも役立ちます。

Lakeflow 宣言型パイプラインのJSON構成設定の詳細については、「Lakeflow宣言型パイプラインの構成」を参照してください。

新しいパイプラインを構成する

新しいパイプラインを構成するには、次の手順を実行します。

1. サイドバーの上部にある 新規 を選択し、 ETL パイプライン 。

2. 上部に、パイプラインに一意の名前を付けます。

3. 名前の下に、選択されたデフォルトのカタログとスキーマが表示されます。これらを変更して、パイプラインに異なるデフォルトを設定します。

   デフォルトのカタログとデフォルトのスキーマは、コード内でカタログまたはスキーマを使用してデータセットを修飾していない場合に、データセットの読み取りまたは書き込みが行われる場所です。詳細については、 Databricksのデータベース オブジェクト」を参照してください。

4. パイプラインを作成するには、希望するオプションを選択します。

   • SQL のサンプル コードから開始して、 SQL のサンプル コードを含む新しいパイプラインとフォルダー構造を作成します。
   • Python のサンプル コードから始めて、 Python のサンプル コードを含む新しいパイプラインとフォルダー構造を作成します。
   • 単一の変換から始めて、 新しい空のコード ファイルを使用して、新しいパイプラインとフォルダー構造を作成します。
   • 既存のアセットを追加して 、ワークスペース内の既存のコード ファイルに関連付けることができるパイプラインを作成します。
   • ソース管理されたプロジェクトを作成して、 新しい Databricks Asset Bundles プロジェクトでパイプラインを作成するか、既存のバンドルにパイプラインを追加します。

   ETL パイプラインには、SQL と Python の両方のソース コード ファイルを含めることができます。新しいパイプラインを作成し、サンプル コードの言語を選択すると、その言語はデフォルトでパイプラインに含まれるサンプル コードのみに適用されます。

5. 選択すると、新しく作成されたパイプラインにリダイレクトされます。

   ETL パイプラインは、次のデフォルト設定で作成されます。

   • Unity Catalog
   • 現在のチャンネル
   • サーバーレスコンピュート
   • 開発モードはオフです。この設定は、パイプラインのスケジュールされた実行にのみ影響します。エディターからパイプラインを実行すると、常にデフォルトで開発モードが使用されます。

   この構成は、開発やテストなどの多くのユースケースに推奨されており、スケジュールに従って実行する必要がある本番運用ワークロードに適しています。 パイプラインのスケジュール設定の詳細については、 「ジョブのパイプライン タスク」を参照してください。

   これらの設定はパイプライン ツールバーから調整できます。

あるいは、ワークスペース ブラウザから ETL パイプラインを作成することもできます。

1. 左側のパネルで ワークスペース をクリックします。
2. Git フォルダーを含む任意のフォルダーを選択します。
3. 右上隅の [作成] をクリックし、 [ETL パイプライン] をクリックします。

[ジョブとパイプライン] ページからETLパイプラインを作成することもできます。

1. ワークスペースで、サイドバーの ジョブ & パイプライン をクリックします。
2. [新規] の下で、 [ETL パイプライン] をクリックします。

コンピュート構成オプション

Databricksでは常に 強化 オートスケール を使用することをお勧めします。 他のコンピュート構成のデフォルト値は、多くのパイプラインで適切に機能します。

サーバレス パイプラインではコンピュート設定オプションが削除されます。サーバレス パイプラインの設定手順については、 サーバレス パイプラインの設定を参照してください。

コンピュート構成をカスタマイズするには、次の設定を使用します。

• ワークスペース管理者は、 クラスターポリシー を構成できます。 コンピュート ポリシー 管理者は、ユーザーが使用できるコンピュート オプションを制御できます。 コンピュートポリシーの選択を参照してください。

• オプションで、 クラスターモード を 固定サイズ または レガシーオートスケール で実行するように構成できます。 「オートスケールを使用した Lakeflow 宣言型パイプラインのクラスタリング使用率の最適化」を参照してください。

• オートスケールが有効になっているワークロードの場合は、 最小ワーカー と 最大ワーカー を設定して、スケーリング動作の制限を設定します。 「 Lakeflow宣言型パイプラインにおけるクラシックコンピュートの設定」を参照してください。

• オプションでPhotonアクセラレーションをオフにすることができます。 Photonとはを参照してください。

• パイプラインがインスタンスプロファイルを使用してクラウドサービスへのアクセスを制御する場合は、 インスタンスプロファイル を選択します。 クラウドストレージの設定を参照してください。

• クラスタータグ を使用して、LakeFlow宣言型パイプラインに関連するコスト監視します。コンピュートタグの設定を参照してください。

• インスタンスタイプ を設定して、パイプラインの実行に使用する仮想マシンのタイプを指定します。パイプラインを実行するためのインスタンスタイプを選択するを参照してください。

  • パイプラインで構成されたワークロードに最適化された ワーカー タイプ を選択します。
  • オプションで、ワーカー タイプとは異なる ドライバー タイプ を選択することもできます。 これは、大規模なワーカー タイプと低いドライバー コンピュート使用率を使用するパイプラインのコストを削減したり、多数の小規模なワーカーを含むワークロードでのメモリ不足の問題を回避するためにより大きなドライバー タイプを選択したりする場合に役立ちます。

実行ユーザーを設定する

実行ユーザーを使用すると、パイプラインが実行に使用する ID と、パイプラインが作成または更新するテーブルの所有権を変更できます。 これは、パイプラインを作成した元のユーザーが非アクティブ化された場合 (たとえば、会社を辞めた場合) に役立ちます。このような場合、パイプラインが機能しなくなり、パブリッシュされたテーブルが他のユーザーがアクセスできなくなる可能性があります。パイプラインを更新して、別の ID (サービスプリンシパルなど) として実行し、パブリッシュされたテーブルの所有権を再割り当てすることで、アクセスを復元し、パイプラインが引き続き機能するようにすることができます。 サービスプリンシパルとして実行されるパイプラインは、個々のユーザーに縛られず、自動化されたワークロードに対してより安全、安定性、信頼性が高まるため、ベスト プラクティスと見なされます。

必要な権限

変更を行うユーザーの場合:

• パイプラインに対する CAN_MANAGE 権限
• サービスプリンシパルの CAN_USE ロール (実行-as をサービスプリンシパルに設定する場合)

実行-as ユーザーまたはサービスプリンシパルの場合:

• ワークスペース アクセス:

  • ワークスペース内で操作するためのワークスペースアクセス 権限
  • パイプラインで使用するクラスターポリシーの権限 を使用できます
  • ワークスペースでのコンピュート作成許可

• ソースコードアクセス:

  • パイプラインのソースコードに含まれるすべてのノートブックの 読み取り権限を持つ
  • パイプラインがワークスペースファイルを使用する場合、そのファイルに対する 読み取り権限を持つ

• Unity Catalog権限 ( Unity Catalogを使用するパイプラインの場合):

  • USE CATALOG 対象カタログ
  • USE SCHEMA ターゲットスキーマのCREATE TABLE
  • MODIFY パイプラインが更新する既存のテーブルに対する権限
  • CREATE SCHEMA パイプラインが新しいスキーマを作成する場合の権限

• レガシーHive metastoreアクセス許可 ( Hive metastoreを使用するパイプラインの場合):

  • SELECT およびターゲットデータベースとテーブルに対するMODIFY権限

• 追加のクラウド ストレージ アクセス (該当する場合):

  • ソースストレージの場所から読み取る権限
  • ターゲットストレージの場所への書き込み権限

実行ユーザーの設定方法

run-asユーザーは、パイプライン モニタリング ページまたはパイプライン エディターのパイプライン設定を通じて設定できます。 パイプラインモニタリングページからユーザーを変更するには:

1. [ジョブとパイプライン] をクリックしてパイプラインのリストを開き、編集するパイプラインの名前を選択します。

2. パイプラインのモニタリング ページで、 [設定] をクリックします。

3. パイプライン設定 サイドバーで、 [実行者として] の横にある[編集] をクリックします。

4. 編集ウィジェットで、次のいずれかのオプションを選択します。

   • あなた自身のユーザーアカウント
   • CAN_USE 権限を持つサービスプリンシパル

5. 変更を適用するには、 「保存」 をクリックします。

実行ユーザーを正常に更新すると、次のようになります。

• パイプライン ID は、今後のすべての実行で新しいユーザーまたはサービスプリンシパルを使用するように変更されます。
• Unity Catalogパイプラインでは、パイプラインによって公開されたテーブルの所有者が、新しい実行-as ID に一致するように更新されます。
• 今後のパイプラインの更新では、新しい実行-の権限と資格情報が ID として使用されます。
• 継続的なパイプラインは新しい ID で自動的に再起動します。トリガーされたパイプラインは自動的に再起動せず、実行-as の変更によりアクティブな更新が中断される可能性があります

注記

実行-as の更新が失敗した場合、失敗の理由を説明するエラー メッセージが表示されます。 一般的な問題には、サービスプリンシパルに対する権限が不十分であることが含まれます。

その他の構成上の考慮事項

パイプラインでは次の構成オプションも使用できます。

• Advanced 製品エディションでは、Lakeflow 宣言型パイプラインのすべての機能にアクセスできます。必要に応じて、 Pro または Core 製品エディションを使用してパイプラインを実行できます。「製品エディションの選択」を参照してください。

• 本番運用でパイプラインを実行する場合は、 継続的 パイプライン モードを使用することを選択できます。 トリガー パイプライン モードと継続的パイプライン モードを参照してください。

• ワークスペースがUnity Catalog 用に設定されていない場合、またはワークロードでレガシーHive metastoreを使用する必要がある場合は、「レガシーHive metastoreでLakeflow宣言型パイプラインを使用する 」を参照してください。

• 成功または失敗の条件に基づくEメール更新の 通知 を追加します。 パイプラインイベントのEメール通知の追加を参照してください。

• 設定 フィールドを使用して、パイプラインのキーと値のペアを設定します。これらの構成には、次の 2 つの目的があります。

  • ソース コードで参照できる任意のパラメーターを設定します。 「Lakeflow宣言型パイプラインでパラメーターを使用する」を参照してください。
  • パイプライン設定と Spark 設定を構成します。Lakeflow宣言型パイプラインのプロパティ リファレンスを参照してください。
  • タグ を設定します。タグは、ワークフロー リストに表示されるパイプラインのキーと値のペアです。パイプライン タグは課金に関連付けられていません。

• Preview チャンネルを使用して、保留中の Lakeflow 宣言型パイプライン ランタイムの変更に対してパイプラインをテストし、新機能を試用します。

製品エディションを選択してください

Lakeflow 宣言型パイプライン 製品エディションを選択し、パイプライン要件に最適な機能を備えています。次の製品エディションが利用可能です。

• Core を使用して、ストリーミング取り込みワークロードを実行します。パイプラインがチェンジデータキャプチャ (CDC) や Lakeflow 宣言型パイプラインの期待値などの高度な機能を必要としない場合は、Coreエディションを選択します。
• Pro ストリーミング取り込みと CDC ワークロードを実行します。Pro製品エディションは、 Coreのすべての機能をサポートするほか、ソース データの変更に基づいてテーブルを更新する必要があるワークロードもサポートします。
• Advanced ストリーミング取り込みワークロード、CDC ワークロード、およびエクスペクテーションが必要なワークロードを実行します。Advanced 製品エディションは、Core エディションと Pro エディションの機能をサポートし、Lakeflow宣言型パイプラインの期待値によるデータ品質制約が含まれています。

製品エディションは、パイプラインを作成または編集するときに選択できます。パイプラインごとに異なるエディションを選択できます。Lakeflow 宣言型パイプライン 製品ページを参照してください。

注: パイプラインに、エクスペクテーションなど、選択した製品エディションでサポートされていない機能が含まれている場合は、エラーの理由を説明するエラー メッセージが表示されます。 その後、パイプラインを編集して適切なエディションを選択できます。

ソースコードを構成する

LakeFlow Pipelines Editor のアセットブラウザを使用して、パイプラインを定義するソースコードを設定できます。 パイプラインのソース コードは、ワークスペース ファイルに保存されている SQL または Python スクリプトで定義されます。パイプラインを作成または編集するときに、1 つ以上のファイルを追加できます。デフォルトでは、パイプラインのソース コードは、パイプラインのルート フォルダー内のtransformationsフォルダーにあります。

Lakeflow宣言型パイプラインはデータセットの依存関係を自動的に分析してパイプラインの処理グラフを作成するため、ソース コード アセットを任意の順序で追加できます。

LakeFlow Pipelinesエディターの使用の詳細については、 「 LakeFlow Pipelinesエディターを使用したETLパイプラインの開発とデバッグ」を参照してください。

Python を使用するパイプラインの外部依存関係を管理する

Lakeflow 宣言型パイプラインでは、 Python パッケージやライブラリなど、パイプラインでの外部依存関係の使用がサポートされています。 依存関係の使用に関するオプションと推奨事項については、「Lakeflow宣言型パイプラインの Python依存関係の管理 」を参照してください。

Databricksワークスペースに保存されているPythonモジュールを使用する

パイプラインのソース コード ファイルに Python コードを実装するだけでなく、Databricks Git フォルダーまたはワークスペース ファイルを使用して、コードを Python モジュールとして保存することもできます。コードをPythonモジュールとして保存することは、同じパイプライン内の複数のパイプラインまたはノートブックで使用したい共通の機能がある場合に特に便利です。 パイプラインで Python モジュールを使用する方法については、 「Git フォルダーまたはワークスペース ファイルから Python モジュールをインポートする」を参照してください。