Lakeflow Jobsとは？Databricksのワークフロー管理

Lakeflow Jobsは、Databricks上でデータパイプラインやMLワークフローをスケジュール実行・監視するためのオーケストレーション基盤です。 旧称「Databricks Workflows」「Databricks Jobs」が2024年後半に「Lakeflow Jobs」へ改称されました。 複数のタスクをDAG（有向非巡回グラフ）で構成し、依存関係・リトライ・パラメータ受け渡しを一元管理します。

この記事では、マルチタスクジョブの構造からスケジューリング、エラーハンドリング、監視までを実務と試験の両面から整理します。

マルチタスクジョブの構造（DAG）

Lakeflow Jobsの基本単位は「ジョブ」で、1つのジョブは1つ以上の「タスク」で構成されます。 タスク間に依存関係を定義するとDAGが形成され、依存先タスクの完了後に次のタスクが自動起動します。

[Ingest] ──→ [Transform_Bronze] ──→ [Transform_Silver] ──→ [Load_Gold]
                    │                                          │
                    └──→ [Data_Quality_Check] ────────────────→┘
                                                               │
                                                          [Notify]

上記のDAGでは、Ingestの完了後にTransform_BronzeとData_Quality_Checkが並列起動し、 Transform_Silver → Load_Goldは直列に実行されます。NotifyタスクはLoad_GoldとData_Quality_Checkの両方が完了してから実行されます。 このように、直列・並列・合流を組み合わせて複雑なパイプラインを表現できます。

依存関係の定義はUI上でドラッグ＆ドロップするか、JSON/YAML定義で depends_on にタスクキーの配列を指定します。 循環依存はバリデーションエラーになります。

タスクタイプ一覧

1つのジョブ内で異なるタスクタイプを混在させることができます。 例えばNotebookでETLを行い、その結果をSQLタスクで集計し、DLTパイプラインでストリーミング処理を実行する構成が1ジョブで完結します。

スケジューリング

Lakeflow Jobsは3種類のトリガー方式をサポートしています。

CRON式スケジュール

標準的な時刻ベースのスケジュールです。タイムゾーンを明示的に指定でき、 例えば 0 0 2 * * ?（毎日午前2時）や 0 0 */6 * * ?（6時間おき）のように設定します。 Quartz CRON形式（秒フィールドあり・6フィールド）を使用するため、 Linux標準の5フィールドCRONとはフィールド数が異なる点に注意が必要です。

ファイル到着トリガー

Unity Catalogで管理されたクラウドストレージのパスを監視し、新しいファイルが到着したタイミングでジョブを起動します。 ポーリングベースではなくイベント駆動に近い挙動で、S3/ADLS/GCSのパスを指定できます。 バッチ取り込みパイプラインで「ファイルが来たら処理する」というパターンに最適です。

連続実行（Continuous）

ジョブが完了するたびに即座に再実行するモードです。ストリーミング処理のように常時稼働させたいNotebookに使います。 再起動間の最小間隔を秒単位で指定でき、失敗時の自動リトライと組み合わせて常時稼働パイプラインを構成します。

リトライとエラーハンドリング

Lakeflow Jobsではリトライをタスクレベルとジョブレベルの2段階で設定できます。

タスクレベルのリトライが設定されている場合、そのタスクが失敗するとまずタスク単体が再実行されます。 max_retriesを超えてもタスクが成功しなければ、そのタスクは最終的に「失敗」となり、 依存する下流タスクはスキップされます。 ジョブ全体の実行結果は「失敗」になりますが、失敗タスクに依存しないブランチのタスクは正常に実行されます。

「修復実行（Repair Run）」機能を使うと、失敗したタスクとそのdown streamタスクだけを再実行できます。 成功済みのタスクは再実行されないため、大規模パイプラインの部分障害回復に有用です。

Job ClusterとAll-Purpose Clusterの使い分け

Lakeflow Jobsのタスクは、Job Cluster（ジョブ専用クラスタ）またはAll-Purpose Cluster（対話型クラスタ）のいずれかで実行します。

本番パイプラインでは原則としてJob Clusterを使います。DBU単価が安く、実行後にクラスタが自動削除されるため リソースの放置リスクがありません。開発フェーズでは、すでに起動中のAll-Purpose Clusterで手動テスト実行し、 本番デプロイ時にJob Cluster設定へ切り替えるのが一般的なワークフローです。

パラメータとタスク値の受け渡し

Lakeflow Jobsでは、外部からジョブに渡す「Job Parameters」と、 タスク間で実行時にデータを受け渡す「Task Values」の2つの仕組みがあります。

Job Parameters

ジョブ定義時にパラメータのキーとデフォルト値を設定し、スケジュール実行やAPI呼び出し時に値を上書きできます。 動的参照として {{job.start_time}}、{{job.run_id}} などの組み込み変数が使えます。 Notebook内からは dbutils.widgets.get("param_name") で取得します。

Task Values（dbutils.jobs.taskValues）

あるタスクの実行結果を後続タスクに渡す仕組みです。

# タスクAで値をセット
dbutils.jobs.taskValues.set(key="row_count", value=df.count())
dbutils.jobs.taskValues.set(key="target_date", value="2026-03-27")

# タスクB（タスクAに依存）で値を取得
row_count = dbutils.jobs.taskValues.get(
    taskKey="task_a", key="row_count"
)
target_date = dbutils.jobs.taskValues.get(
    taskKey="task_a", key="target_date"
)

値はJSON互換の型（文字列、数値、ブール値）で渡せます。 大きなDataFrameやバイナリデータは渡せないため、そのような場合はDelta Tableや一時ファイルに書き出して パスのみをtaskValueで共有するパターンを使います。

通知設定

ジョブの開始・成功・失敗・スキップなどのイベントに対して通知を設定できます。

• メール通知: 任意のメールアドレスに送信。複数アドレス指定可能
• Webhook: 任意のHTTPSエンドポイントにPOSTリクエストを送信。PagerDutyやカスタムシステムと連携
• Slack連携: Databricksの通知先としてSlack Webhookを登録し、チャンネルへ自動投稿
• Microsoft Teams: Incoming Webhook経由で通知

通知はジョブレベルとタスクレベルの両方で設定可能です。 例えば「ジョブ全体の失敗はオンコールチームに通知」「特定タスクの失敗はデータ品質チームに通知」のように 粒度を分けることで、アラート疲れを防ぎつつ適切な担当者に情報を届けられます。

監視（実行履歴とSystem Tables）

Lakeflow JobsのUIから各ジョブの実行履歴を確認でき、成功・失敗・実行時間・各タスクのステータスが一覧表示されます。 過去60日間の実行履歴がデフォルトで保持されます。

より高度な分析や長期保持には、Unity CatalogのSystem Tablesを使います。system.lakeflow スキーマに以下のテーブルが用意されています。

これらのテーブルをSQLで集計すると、「過去30日で最も失敗率が高いジョブ」「平均実行時間が増加傾向にあるタスク」 「特定時間帯のクラスタ利用集中」などを可視化できます。 Databricks SQLダッシュボードと組み合わせると、運用チーム向けのジョブ健全性モニタリングを構築できます。

試験で問われるポイント

Data Engineer Associate試験では「Production Pipelines」ドメイン（約16%）でLakeflow Jobsが出題されます。 以下のパターンが頻出です。

• 本番パイプラインでJob ClusterとAll-Purpose Clusterのどちらを使うべきか → Job Cluster（コスト・自動終了の観点）
• ジョブが失敗したときに失敗タスクだけを再実行する方法 → 修復実行（Repair Run）
• 前のタスクの計算結果を後のタスクに渡す方法 → dbutils.jobs.taskValues.set() / get()
• ファイル到着時にジョブを起動する設定 → ファイル到着トリガー
• CRON式のフィールド数 → Quartz形式の6フィールド（秒を含む）
• タスクの依存関係に循環がある場合 → バリデーションエラー（DAGでなくなるため作成不可）

試験では「DLT pipeline」「Lakeflow Jobs」「Notebook単独実行」の3つを混同させる選択肢が多く、 それぞれの目的と適用場面の違いを整理しておくことが重要です。

問題で確認

よくある質問