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

Lakeflow Jobsは、Databricks上でデータパイプラインやMLワークフローをスケジュール実行・監視するためのオーケストレーション基盤です。 旧称は「Databricks Jobs」（左メニューでは「Workflows」）で、2025年6月11日に「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のDepends on欄で上流タスクを選ぶか、JSON/YAML定義の depends_on に上流タスクのtask_keyを並べます。 循環依存はバリデーションエラーになります。

タスクタイプ一覧

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

スケジューリング

Lakeflow Jobsを自動で起動するトリガーには、スケジュール・テーブル更新・ファイル到着・モデル更新（ベータ）・連続実行があります。ほかに手動（Run now）や外部のオーケストレーションツールから起動することもできます。DEA試験ガイドが挙げるのはスケジュール・ファイル到着・テーブル更新の3つです。ここではその3つと連続実行を説明します。

CRON式スケジュール

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

ファイル到着トリガー

Unity Catalogの外部ロケーションまたはボリューム（その配下のパスも可）を監視し、新しいファイルが到着したらジョブを起動します。S3/ADLS/GCS上の場所を対象にできます。 新着ファイルの確認は1分ごとのベストエフォートで、外部ロケーションでファイルイベントを有効にすると、クラウドの変更通知を使って効率よく検知できます。 ファイルの到着時刻が不規則で、「ファイルが来たら処理する」というバッチ取り込みパイプラインに最適です。

テーブル更新トリガー

監視しているテーブルが更新（更新・マージ・削除を含む）されたらジョブを起動します。 Unity Catalogのマネージドテーブル（DeltaとIceberg）、Delta形式の外部テーブル、マテリアライズドビュー、ストリーミングテーブル、それらに依存するビューを1つ以上指定でき、いずれか1つが更新されたときと、すべてが更新されたときのどちらで起動するかを選べます。詳細設定の「トリガー間の最小時間」と「最後の変更後の待機時間」を使うと、続けて届く更新を1回の実行にまとめられます。 ジョブの実行ユーザー（Run as）には、監視するテーブルへのSELECTと、親のカタログ・スキーマへのUSE CATALOG・USE SCHEMAが必要です。

連続実行（Continuous）

常に1つの実行が動き続けるモードで、実行が終わると（成功・失敗を問わず）60秒未満の遅延の後に次の実行が始まります。常時稼働のストリーミング処理に使います。 連続実行のジョブにはリトライポリシーを設定できず、失敗時はジョブ全体が指数バックオフで自動的に再試行されます。実行中に失敗したタスクの再試行はTask retry modeで指定します（既定はOn failure）。

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

リトライポリシーはタスクごとに設定します。多くの構成では既定でリトライしませんが、サーバーレスのジョブは既定の自動最適化で失敗したタスクを再試行し、連続実行のジョブは前述の指数バックオフで再試行します。下の表の最後の行はリトライではなく、ジョブ単位の同時実行の設定です。

タスクにリトライが設定されている場合、そのタスクが失敗するとまずタスク単体が再実行されます。 max_retriesを超えてもタスクが成功しなければ、そのタスクは最終的に「失敗」となり、 既定のRun if（All succeeded）の下流タスクは「Upstream failed」の状態になって実行されません。 ジョブ全体の実行結果は「失敗」になりますが、失敗タスクに依存しないブランチのタスクは正常に実行されます。

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

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

Lakeflow Jobsのタスクを動かすコンピュートには、サーバーレスのジョブコンピュート（対応するタスクでは既定）、クラシックのJob Cluster（ジョブ専用クラスタ）、SQLタスク用のSQLウェアハウス、パイプラインのコンピュートがあります。All-Purpose Cluster（対話型クラスタ）も選べますが、本番ジョブには推奨されていません。ここではクラシックの2つを比べます。

本番パイプラインでは原則としてジョブ用のコンピュート（サーバーレスまたはJob Cluster）を使います。Job ClusterはAll-PurposeよりDBU単価が低く、実行後にクラスタが自動で終了するため リソースの放置リスクがありません。開発フェーズでは、すでに起動中のAll-Purpose Clusterで手動テスト実行し、 本番デプロイ時にジョブ用のコンピュートへ切り替えるのが一般的なワークフローです。

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

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

Job Parameters

ジョブ定義時にパラメータのキーとデフォルト値（固定値または動的参照）を設定し、UIやAPIから値を変えて実行するときや、修復実行のときに値を上書きできます。 動的参照として {{job.start_time.iso_date}}、{{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で表せるもの（文字列、数値、ブール値、リストなど）に限られ、JSONにしたときの大きさは48 KiBまでです。 大きなDataFrameやバイナリデータは渡せないため、そのような場合はDelta Tableや一時ファイルに書き出して パスのみをtaskValueで共有するパターンを使います。

通知設定

ジョブの開始・成功・失敗・実行時間のしきい値超過・ストリーミングのバックログ超過といったイベントに対して通知を設定できます。スキップやキャンセルされた実行の通知は止める（ミュートする）こともできます。メール以外の通知先は、管理者があらかじめシステムの通知先として登録しておきます。

• メール通知: 任意のメールアドレスに送信。複数アドレス指定可能
• Webhook: 任意のHTTPSエンドポイントに送信し、自社システムなどと連携
• Slack連携: Databricksの通知先としてSlack Webhookを登録し、チャンネルへ自動投稿
• Microsoft Teams: WebhookのURLを通知先として登録して通知
• PagerDuty: PagerDutyの統合キーを通知先として登録して通知

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

監視（実行履歴とSystem Tables）

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

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

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

試験で問われるポイント

Data Engineer Associate試験では、Lakeflow Jobsは「Working with Lakeflow Jobs」（16%）という独立したセクションで出題されます。 ガイドの目標に沿って、次の論点を押さえておきましょう。

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

「Lakeflow Spark Declarative Pipelinesのパイプライン」「Lakeflow Jobs」「Notebook単独実行」の役割の違いも整理しておきましょう。 パイプラインはデータセットと変換の関係を宣言的に定義し、ジョブはタスクの実行順序と起動条件を手続き的に定義します。パイプラインはジョブの中の1つのタスクとして実行できます。

問題で確認

よくある質問