Lakeflow ジョブは、タスク間のリレーションシップを定義するための手続き型のアプローチを提供 します。 Lakeflow パイプラインは、 データセット と変換の間のリレーションシップを定義するための宣言型アプローチ を提供します。 ジョブUI、LakeflowパイプラインUI、またはSQLを使って、ジョブ内のタスクとしてパイプラインをスケジューリングします。
パイプラインタスクは、ジョブのスケジュールに応じて2つの方法でパイプラインを実行します。
- トリガーまたはスケジュールされたジョブでは、パイプラインタスクは単一の更新を開始し、更新が完了すると停止します。
- 連続ジョブでは、パイプラインタスクがパイプラインを連続的に実行します。 ジョブのスケジュールが実行モードを決定するため、パイプラインは自身の パイプラインモード 設定がトリガーされていても連続的に実行されます。 「 連続ジョブでパイプラインを連続的に実行」を参照してください。
トリガーされたパイプラインと継続的パイプラインの詳細については、 トリガーされたパイプラインと継続的パイプラインモードを参照してください。
ジョブ UI を使用してパイプライン タスクを構成する
Lakeflow パイプラインは、パイプライン定義でソース コードとコンピューティングのすべての構成を管理します。
ジョブにパイプラインを追加するには、次の手順を実行します。
- 新しいタスクを作成して名前を付け、種類のパイプラインを選択します。
- [ パイプライン ] ドロップダウン メニューで、既存のパイプラインを選択します。
- 必要に応じて、パイプラインで完全更新をトリガーできます。
- 必要に応じて、[ パラメーター] フィールド でパラメーターのオーバーライドを設定できます。 「パラメーター」を参照してください。
- (省略可能)再試行、実行期間、ストリーミング バックログのしきい値、または通知を構成するには、「 タスクの詳細設定」を参照してください。
Note
を選択して、タスクの作成時に新しいインジェスト パイプラインを作成することもできます。[作業ウィンドウの追加] または [タスクの種類] ドロップダウンからの新しいインジェスト パイプライン。
このタスクを編集、複製、無効化、または削除するには、「 Lakeflow ジョブのタスクの構成と編集」を参照してください。
連続ジョブでパイプラインを連続的に実行します
連続ジョブにパイプラインタスクが含まれている場合、ジョブはパイプラインを連続的に実行します。 パイプラインの内蔵パイプライン モード を連続に設定する必要はありません。ジョブのスケジュールが実行モードを決定し、 パイプラインモード 設定よりも優先されます。
これはLakeflowパイプラインにのみ適用されます。 スタンドアロンのマテリアライズされたビューやストリーミングテーブルは常にトリガーモードで動作します。
連続ジョブでラップされた連続パイプラインは、標準モードなどのサーバーレス パフォーマンスモードを利用できますが、パイプラインの組み込み連続モードはこれをサポートしていません。
Databricks では、パイプラインに組み込まれている継続設定ではなく、継続ジョブを使用して継続パイプラインを実行することを推奨しています。 予期しない動作を避けるために、パイプラインを連続ジョブでラップする際のパイプライン モード をトリガーモード(デフォルト)に設定してください。 詳細は 「連続ジョブで連続パイプラインを実行する」をご覧ください。
連続ジョブはジョブUI内、パイプラインページから直接、または宣言的自動化バンドルで作成できます。 「 パイプラインUIでパイプラインをスケジューリングする 」および「宣言 的自動化バンドルで連続パイプラインジョブを定義する」を参照してください。
宣言的自動化バンドルで連続パイプラインジョブを定義します
以下の宣言的自動化バンドルの例は、パイプラインを実行する連続ジョブをタスクとして定義しています。
continuous.pause_statusをUNPAUSEDに設定するとパイプラインが継続的に実行され、performance_target: STANDARD標準パフォーマンスモードでジョブを実行します。
# resources/continuous_job.yml
resources:
jobs:
continuous_pipeline_job:
name: continuous_pipeline_job
performance_target: STANDARD
continuous:
pause_status: UNPAUSED
email_notifications:
on_failure:
- your_email@example.com
tasks:
- task_key: refresh_pipeline
pipeline_task:
pipeline_id: ${resources.pipelines.example_pipeline.id}
既存の連続パイプラインを連続ジョブに移行するには、パイプライン定義から continuous フィールドを削除してください。 ジョブは連続実行を設定します。
同等のJobs APIペイロードについては、 jobs/create 参照を参照してください。
データベース テーブル同期パイプライン
データベース テーブル同期パイプライン タスクは、Lakebase 同期テーブルを維持するパイプラインを実行するパイプライン タスクです。 これを使用して、同期されたテーブルをスケジュールに従って更新するか、ソース Unity カタログ テーブルが変更されたときに、運用アプリケーションが Lakebase Postgres から現在のデータを読み取るようにします。
タスクの [種類] ドロップダウンに、[インジェストと変換] の下にデータベース テーブル同期パイプラインが表示されます。 これを選択すると、パイプライン タスクが構成されます。 [ パイプライン ] フィールドで、更新する同期されたテーブルに関連付けられているパイプラインを選択します。
Lakebase には 2 つの提供形態があります。 更新する同期されたテーブルによって、次の手順のセットが決まります。
- Lakebase オートスケーリング は、新しいインスタンス向けの提供機能です。 自動スケール同期テーブルのタスクを構成するには、「 データベース テーブル同期パイプライン タスク」を参照してください。
- Lakebase Provisioned は従来の提供形態です。 プロビジョニングされた同期テーブルのタスクを構成するには、「 データベース テーブル同期パイプライン タスク」を参照してください。
インジェスト パイプライン
インジェスト パイプライン タスクは、インジェスト パイプラインを実行するパイプライン タスクです。 タスクの [種類] ドロップダウンで、[ インジェスト パイプライン ] を選択すると、 データの追加 ウィザードが開始され、インジェスト パイプラインのパイプライン タスクが作成されます。
ウィザードの最初のページで、データ ソースが要求されます。 次のページは、選択したソースによって異なります。 たとえば、MySQL を選択した場合、手順については、オプション 1: Azure Databricks UI を参照してください。 使用可能なソースについては、「 Lakeflow Connect のマネージド コネクタ」を参照してください。
Parameters
Von Bedeutung
この機能は ベータ版です。 ワークスペース管理者は、[ プレビュー] ページからこの機能へのアクセスを制御できます。 Manage Azure Databricks プレビューを参照してください。
動的値参照を使用して、パイプライン タスクの ジョブ または タスク パラメーターにアクセスできます。 タスク構成のパラメーターに Key-Value ペアを追加することで 、パラメーター をオーバーライドできます。
パイプライン内からパラメーター値にアクセスする方法の詳細については、 タスクからのパラメーター値へのアクセスに関するページを参照してください。
パイプライン タスクでのコンカレンシーの制限
パイプラインは、一度に 1 つの更新のみを実行できます。 パイプライン タスクを含むジョブには、次のコンカレンシー キャップが適用されます。
- パイプライン タスクを含む
max_concurrent_runs > 1を持つジョブは、1 回の同時実行に制限されます。 この上限が適用されると、ジョブ UI に通知が表示されます。 - for-each タスクにラップされたパイプライン タスクは、ループの構成されたコンカレンシーに関係なく、1 つの同時実行イテレーションに制限されます。
多数のパラメーターの組み合わせまたは厳密なスケジュールで実行する予定のパラメーター化されたパイプラインを設計する場合は、これらの上限について計画します。
パイプライン UI を使用してパイプラインをスケジュールする
パイプラインにスケジュールを追加すると、1 つのパイプライン タスクでジョブが作成されます。 トリガーオプションの詳細については、「 ジョブ UI を使用してパイプライン タスクを構成する」を参照してください。
次の手順を実行して、パイプライン UI を使用してスケジュールされたジョブでパイプライン タスクを構成します。
- ワークスペースで、[ワークフロー] アイコンをクリック
サイドバーのジョブとパイプライン。 - パイプライン Name をクリックします。 パイプライン UI が表示されます。
-
[スケジュール] をクリックします。
- パイプラインのスケジュールが存在しない場合は、[ 新しいスケジュール] ダイアログボックスが表示されます。
- 1 つ以上のスケジュールが既に存在する場合は、スケジュールを追加をクリックします。
-
トリガータイプドロップダウンリストでトリガータイプを選択してください:
- Scheduled は、時間ベースのスケジュールを作成します。 「 Advanced 」をクリックすると、cron構文を含む多くのオプションがあります。
- Continuous は、パイプラインを継続的に実行する continuous job を作成します。 「 連続ジョブでパイプラインを連続的に実行」を参照してください。
- ジョブ名フィールド[]に、一意のジョブ名を入力します[]。
- (任意)標準パフォーマンスモードでパイプラインを実行するには、 パフォーマンス最適化 のチェックボックスをクリアしてください。 パフォーマンス モードの選択を参照してください。
- (省略可能)その他のオプションで、パイプラインの開始時、成功、または失敗時にアラートを受信するように 1 つ以上の電子メール アドレスを構成します。
- Create をクリックしてください。
連続スケジュールの場合、Azure Databricksは自動的に実行を開始します。 停止するには、パイプラインページで 「停止 」をクリックするかスケジュールを一時停止してください。 両方のアクションはアクティブなアップデートもキャンセルします。
Note
パイプラインが 1 つ以上のスケジュールされたジョブに含まれている場合、スケジュール ボタンには、スケジュール (5)など、既存のスケジュールの数が表示されます。
Databricks SQL の具体化されたビューまたはストリーミング テーブルにスケジュールを追加する
Databricks SQL で定義されている具体化されたビューとストリーミング テーブルは、CREATE コマンドまたは ALTER コマンドで指定された時間ベースのスケジュール設定をサポートします。
詳細については、次の記事を参照してください。
その他のリソース
- パイプラインタスク:宣言的自動化バンドルでコードとしてパイプラインタスクを定義します。
- トリガーモードと連続パイプラインモード:パイプライントリガーモードについて学びましょう。
- タスクパラメータの設定:タスクにパラメータを渡します。
