ジョブのパイプラインタスク

Lakeflow Jobsは、 タスク 間の関係を定義するための手続き型アプローチを提供します。LakeFlow Pipelines は、 データセット と 変換 の間の関係を定義するための宣言型アプローチを提供します。ジョブUI、LakeFlow Pipelines UI、またはSQLを使用して、ジョブ内のタスクとして実行するようにパイプラインをスケジュールします。

パイプラインタスクは、ジョブのスケジュールに応じて、次の2つのいずれかの方法でパイプラインを実行します：

• In a Triggered or scheduled ジョブでは、パイプラインタスクは単一の更新を起動し、更新が完了すると停止します。
• 連続ジョブでは、パイプラインタスクがパイプラインを継続的に実行します。ジョブのスケジュールによって実行モードが決定されるため、パイプライン自体の パイプラインモード 設定がTriggerであっても、パイプラインは継続的に実行されます。連続ジョブを使用してパイプラインを継続的に実行するを参照してください。

Triggerパイプラインと継続的パイプラインの詳細については、「Triggerパイプラインモードと継続的パイプラインモード」を参照してください。

ジョブ UI でパイプライン タスクを構成する

LakeFlow Pipelines は、パイプライン定義内のソースコードとコンピュートのすべての構成を管理します。

ジョブにパイプラインを追加するには、次のステップを実行します。

1. 新しいタスクを作成して名前を付け、 種類 で パイプライン を選択します。
2. パイプライン ドロップダウン メニューで、既存のパイプラインを選択します。
3. 必要に応じて、パイプラインで完全な更新をトリガーできます。
4. [Parameters] フィールドで、必要に応じてパラメーターの上書きを設定できます。See パラメーター.
5. (オプション) 再試行、実行期間またはストリーミングバックログのしきい値、または通知を設定するには、高度なタスク設定を参照してください。

注記

タスクを作成するときに、**タスクの追加**ペインまたはタスクの**種類**ドロップダウンから**新しいインジェスト パイプライン**を選択して、新しいインジェスト パイプラインを作成することもできます。

このタスクを編集、複製、無効化、または削除するには、「Lakeflowジョブでタスクを構成および編集する」を参照してください。

連続ジョブを使用してパイプラインを継続的に実行する

継続的ジョブにパイプラインタスクが含まれている場合、そのジョブはパイプラインを継続的に実行します。パイプラインの組み込みの パイプラインモード を「継続的」に設定する必要はありません。ジョブのスケジュールが実行モードを決定し、 パイプラインモード の設定よりも優先されます。

これは Lakeflow pipelines にのみ適用されます。スタンドアロンのマテリアライズドビューとストリーミングテーブルは、常にTriggerモードで実行されます。

連続ジョブにラップされた継続的パイプラインは、パイプラインの組み込みの継続的モードではサポートされていない、標準モードなどのServerlessパフォーマンスモードを使用できます。

Databricksでは、パイプラインの組み込みの継続的設定を使用するのではなく、連続ジョブを使用して継続的パイプラインを実行することを推奨しています。予期しない動作を避けるため、パイプラインを連続ジョブにラップする場合は、パイプラインの パイプラインモード をTrigger（default）に設定してください。詳細については、「連続ジョブを使用して継続的パイプラインをランする」を参照してください。

継続的ジョブは、ジョブUI、パイプラインページから直接、または宣言型オートメーションバンドルを使用して作成できます。パイプラインUIを使用したパイプラインのスケジュールおよび宣言型オートメーションバンドルでの継続的パイプラインジョブの定義を参照してください。

宣言型オートメーションバンドルで継続的なパイプラインジョブを定義する

以下のDeclarative Automation Bundlesの例では、パイプラインをタスクとして実行する連続ジョブを定義しています。continuous.pause_statusをUNPAUSEDに設定するとパイプラインが継続的に実行され、performance_target: STANDARDはジョブを標準パフォーマンスモードで実行します。

YAML

# 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 フィールドを削除します。その後、ジョブは継続的な実行を構成します。

同等のジョブAPIペイロードについては、「jobs/create」 API リファレンスを参照してください。

データベーステーブル同期パイプライン

データベーステーブル同期パイプライン タスクは、Lakebase同期テーブルを維持するパイプラインを実行するパイプラインタスクです。運用アプリケーションがLakebase Postgresから最新のデータを読み取れるように、スケジュールに基づいて、またはソースのUnity Catalogテーブルが変更されたときに、同期されたテーブルを更新するために使用します。

タスクの**タイプ**ドロップダウンで、**取り込みと変換**の下に**データベーステーブル同期パイプライン**が表示されます。これを選択すると、パイプラインタスクが構成されます。**パイプライン**フィールドで、更新したい同期されたテーブルに関連付けられたパイプラインを選択します。

Lakebase には 2 つのオファリングがあります。更新する同期済みテーブルによって、どの指示に従うべきかが決まります：

• Lakebase オートスケール は、新しいインスタンス向けのオファリングです。オートスケール同期テーブルのタスクを設定するには、データベーステーブル同期パイプラインのタスクを参照してください。
• Lakebase Provisioned は、オリジナルのオファリングです。プロビジョニングされた同期テーブルのタスクを設定するには、データベーステーブル同期パイプラインのタスクを参照してください。

取り込みパイプライン

インジェスト パイプライン タスクは、インジェスト パイプラインを実行するパイプライン タスクです。「タスクの タイプ 」ドロップダウンで、 インジェスト パイプライン を選択すると、 [データの追加] ウィザードが起動し、インジェスト パイプラインのパイプライン タスクが作成されます。

ウィザードの最初のページでデータソースが求められます。表示されるページは、選択したソースによって異なります。たとえば、MySQL を選択した場合は、手順についてオプション 1: Databricks UI を参照してください。利用可能なソースについては、LakeFlow Connect のマネージド コネクタを参照してください。

パラメーター

備考

ベータ版

この機能はベータ版です。ワークスペース管理者は、 プレビュー ページからこの機能へのアクセスを制御できます。Databricksのプレビューを管理するを参照してください。

動的な値参照を使用して、パイプラインタスクでジョブまたはタスクパラメーターにアクセスできます。タスク設定の パラメーター で キー と 値 のペアを追加して、パラメーターを上書きできます。

パイプライン内からパラメーター値にアクセスする方法の詳細については、「タスクからのパラメーター値へのアクセス」を参照してください。

パイプラインタスクの同時実行制限

パイプラインは一度に 1 つの更新のみを実行できます。パイプライン タスクを含むジョブには、次の同時実行上限が適用されます:

• パイプラインタスクが含まれているmax_concurrent_runs > 1のジョブは、単一の並列実行に制限されます。この上限が適用されると、ジョブUIに通知が表示されます。
• 「For Eachタスク」でラップされたパイプラインタスクは、ループで構成された並列実行数に関係なく、1つの並列イテレーションに制限されます。

多数のパラメーターの組み合わせ、または厳しいスケジュールで実行を予定しているパラメーター化されたパイプラインを設計する際は、これらの上限を考慮して計画してください。

パイプラインUIでパイプラインをスケジュールする

パイプラインにスケジュールを追加すると、単一のパイプラインタスクを持つジョブが作成されます。より高度なTriggerオプションについては、「ジョブUIを使用したパイプラインタスクの構成」を参照してください。

次のステップを完了することにより、パイプラインUIを使用してスケジュールされたジョブでパイプラインタスクを構成します。

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

2. パイプラインの 名前 をクリックします。パイプライン UI が表示されます。

3. スケジュール をクリックします。

   • パイプラインのスケジュールが存在しない場合、**新しいスケジュール**ダイアログが表示されます。
   • 1 つ以上のスケジュールが既に存在する場合は、 スケジュールの追加 をクリックします。

4. In the Trigger type drop-down list, select a Trigger type:

   • スケジュール済み は、時間ベースのスケジュールを作成します。cron構文を含むその他のオプションについては、 詳細 を選択します。
   • Continuous （連続）は、パイプラインを継続的に実行する連続ジョブを作成します。継続的なジョブを使用してパイプラインを継続的に実行するを参照してください。

5. ジョブ名 フィールドにジョブの一意の名前を入力してください。

6. （オプション）パイプラインを標準パフォーマンスモードで実行するには、 パフォーマンス最適化 チェックボックスをオフにします。パフォーマンスモードの選択を参照してください。

7. (オプション) その他のオプション で、パイプラインの開始、成功、または失敗時にアラートを受け取るように 1 つ以上の Eメール アドレスを構成します。

8. 作成 をクリックします。

継続的なスケジュールの場合は、Databricks が自動的にランを起動します。停止するには、パイプライン ページで 停止 をクリックするか、スケジュールを一時停止します。どちらの操作も、アクティブな更新をキャンセルします。

注記

パイプラインが 1 つ以上のスケジュールされたジョブに含まれている場合、**スケジュール** ボタンには、既存のスケジュール数が表示されます(例: **スケジュール (5)**)。

Databricks SQL のマテリアライズドビューまたはストリーミングテーブルにスケジュールを追加します。

Databricks SQL で定義されているマテリアライズドビューとストリーミング テーブルでは、 CREATE コマンドまたは ALTER コマンドで指定された時間ベースのスケジューリングがサポートされています。

詳細については、次の記事を参照してください。

• CREATE MATERIALIZED VIEW
• CREATE STREAMING TABLE
• ALTER MATERIALIZED VIEW
• ALTER STREAMING TABLE

その他のリソース

• パイプラインタスク：Declarative Automation Bundles を使用してパイプラインタスクをコードとして定義します。
• Triggerパイプラインモードと連続パイプラインモード: パイプラインのTriggerモードについて学習します。
• タスク パラメーターの設定: タスクにパラメーターを渡します。