処理を1つのジョブに束ねる — タスク種別と依存関係の設計

10.1Lakeflow Jobs の全体像とタスク種別 — 何を1つのジョブに束ねられるのか

ノートブックを1本ずつ手で実行しているうちは、どこまで進んだのかが担当者の記憶の中にしかありません。取り込みが途中で落ちたことに気づかないまま変換を走らせ、ほとんど空のテーブルで集計表を上書きしてしまう。夜間バッチでこれが起きると、翌朝の数字が静かに狂います。処理の順番と、前の処理が成功したかどうかの判定を、人間の記憶の外側に置くのがジョブという仕組みです。

Lakeflow Jobs (旧 Databricks Jobs) では、ジョブが入れ物で、タスクが実際の実行単位です。タスクは1つずつ種別を選び、どの計算資源で動かすか、失敗したら何回やり直すかを個別に決めます。ノートブックの末尾で `dbutils.notebook.run()` を呼んで次の処理につなぐ書き方とは、ここが決定的に違います。呼び出し方式では失敗がノートブック1本分としてしか見えず、やり直しは常に先頭からです。処理をタスクに割ってジョブに束ねておけば、失敗した箇所とその下流だけを Repair run で再実行できます。

タスクに割る見返りは、設定を個別に持てることです。1つのタスクは、実行に使うコンピュート、最大再試行回数の `max_retries`、再試行の最小間隔をミリ秒で指定する `min_retry_interval_millis`、タイムアウト時にも再試行するかを決める `retry_on_timeout`、打ち切り時間の `timeout_seconds` を、それぞれ自分の値として持ちます。上流に3回、下流に0回といった設定が普通にでき、`depends_on` によってリトライ設定が継承されることもありません。1本の巨大なノートブックにまとめると、この調整の単位がすべて失われます。

実行時間の制御は2段構えです。`timeout_seconds` (UI の Timeout) を超えたタスクは Timed Out として打ち切られます。一方、所要時間しきい値の Warning は超過した時点でイベントを出すだけで、処理そのものは止めません。長引いたら知らせたいが止めたくはない、という要件はこちらです。なお Timeout と Retries を併用した場合、タイムアウトは各リトライに個別に適用されます。Timeout 30分・Retries 2回なら、各試行がそれぞれ30分の枠を持ちます。

タスク種別は「何を動かすか」と「どのコンピュートが要るか」で選ぶ

種別は18種類あり、1つのジョブの中に違う種別を自由に混在させられます。ノートブックを起点に SQL で集計し、最後にダッシュボードを更新する構成が1ジョブで完結します。dbt プロジェクトのために外部スケジューラを用意する必要はなく、宣言型パイプラインもノートブックでラップせず起動できます。API と YAML で使うキー名まで含めて、よく使うものを並べます。

真っ先に覚えるのは右端の列です。SQL タスクと Dashboard タスクは、サーバーレスまたは Pro の SQL ウェアハウスでしか動きません。Classic ウェアハウスは選べず、ジョブクラスタや汎用クラスタも指定できません。Dashboard タスクを構成するには、対象ダッシュボードへの CAN VIEW 以上の権限も必要です。

Notebook タスクのソースは Workspace か Git provider の2択です。Git を選ぶと相対パスは `/` や `./` で始めず `etl/bronze/ingest.py` のように書き、1つのジョブの全タスクで使えるリモートリポジトリは1つだけです。出力にも上限があり、全セルの合計が30MB、単一セルが8MB を超えると、その実行はキャンセルされ失敗になります。開発中に入れた `display()` を残したまま本番の全量データを流すと、処理そのものは正しくてもここで落ちます。

SQL タスクの実行対象はクエリ、アラート、`.sql` ファイルの3つで、`.sql` にはセミコロン区切りで複数文を書けます。結果は下流から `{{tasks.<task_name>.output.first_row.<column>}}` で先頭行の列値として読めますが、出力は1,000行かつ48KBまで、保持期間は7日です。Dashboard タスクのデータセットパラメータはタスクの入力欄では指定できず、あらかじめダッシュボードのフィルターウィジェットに紐付けておく必要があります。更新完了を知らせる Subscribers は、Dashboards の UI で設定した通知先とは別に管理されます。

Pipeline タスクが起動できるのは triggered モードのパイプラインだけで、continuous のパイプラインは対象外です。Full refresh を有効にすると完全更新として実行されます。1つのパイプラインは同時に1更新しか実行できないため、Pipeline タスクを含むジョブは Maximum concurrent runs を3にしても実質1実行に制限され、For each の中に置いた場合も Concurrency の値にかかわらず1反復ずつ進みます。Run job タスクは Job ドロップダウンから選んだ既存のジョブを起動しますが、3段を超える入れ子と、直接または間接に互いを起動する循環依存はサポートされません。

どこまでを1つのジョブに入れるか

先に物理的な上限を押さえます。数値そのものが問われます。

上限に余裕があっても、何でも束ねてよいわけではありません。判断の軸は失敗したときに一緒にやり直したい範囲かどうかです。取り込みと変換と集計は同じ日次データの運命共同体なので、1つのジョブにまとめて Repair run の対象範囲に収めます。逆に細かく切りすぎるとコンピュートの確保が毎回挟まるので、数十秒で終わる処理を20個に割っても遅くなるだけです。Repair run は同じ実行 ID の中で、失敗したタスクと未実行の下流タスクだけを再実行する機能です。10タスク中2つが失敗しても、成功済みの8つは結果が再利用されます。束ねる範囲がそのまま部分再実行の単位になる、と考えると設計しやすくなります。

なお、1,000という上限は「そこまで入る」という意味であって「そこまで入れてよい」という意味ではありません。数百のタスクを1つのジョブに詰めると実行履歴の一覧は重くなり、個々のタスクまで追えなくなります。層ごと、ドメインごとにジョブを割るほうが運用は軽くなります。

逆に、別チームが所有していて更新頻度も違うテーブルの処理を同じジョブに入れると、相手側の都合で自分のジョブが赤くなり、通知も自分に飛んできます。この場合はジョブを分け、Run job タスクで呼ぶか、上流テーブルを見張る Table update トリガーで起動します。ジョブをまたぐ依存は、依存関係ではなく起動条件で表現するということです。トリガーはジョブ単位の設定でタスクごとには持てないので、毎朝7時に必ず動かしたい処理とファイル到着時だけ動かしたい処理を無理に束ねると、片方に合わせた空振りの実行がコストとして積み上がります。

10.2DAG を組む — depends_on、Run if 6条件と Excluded、If-else 条件タスク、For each

依存を宣言していないタスクは、ジョブが始まった瞬間に全部いっせいに走り出します。「変換タスクが参照するテーブルが見つからない」という失敗の大半は、コードの問題ではなく `depends_on` の書き忘れです。この依存の網は DAG (有向非巡回グラフ。矢印をたどっても元の位置に戻れない形の図) と呼ばれます。もう1つ決めるべきは、上流が失敗したら止めたいのか、失敗しても通知だけは走らせたいのかです。DAG の設計とは、この2つを書き下ろす作業です。

depends_on は後続タスク側に書く

依存は必ず待つ側のタスクに書きます。先行タスクに「次はこれを動かす」と書くフィールドはなく、`precedes` や `wait_for` は存在しません。次の例は、ジョブ定義を YAML ファイルで管理する仕組みである Declarative Automation Bundles での書き方です。

tasks:
  - task_key: ingest
  - task_key: clean_a
    depends_on:
      - task_key: ingest
  - task_key: clean_b
    depends_on:
      - task_key: ingest
  - task_key: publish
    depends_on:
      - task_key: clean_a
      - task_key: clean_b

この定義では `ingest` の成功後に `clean_a` と `clean_b` が並列で起動し、両方が成功してから `publish` が走ります。合流の条件を別に書かなくてよいのは、`depends_on` に2つ並べれば既定の Run if「All succeeded」が両方の成功を要求してくれるからです。ここで押さえるべきは、`tasks` 配列の並び順は画面の表示順を決めるだけで、実行順序には一切関係しない点です。並び順で順序を表現したつもりの定義は、実際には全タスクが同時に起動します。

日本語の要件からの読み替えは機械的です。「A の完了後に B と C を並列で起動し、B の完了後に D を起動」なら、B と C の `depends_on` がそれぞれ `[A]`、D の `depends_on` が `[B]` です。矢印の向きを逆に書くと上流が下流を待つ形になり、永久に起動しない定義ができあがります。粒度は再実行したい単位で割るのが目安です。

Run if の6条件と、満たされなかったときの状態

依存を張っただけのタスクは、Run if の既定値 All succeeded で動きます。上流が1つでも失敗すると下流は Upstream failed になり、実行されません。困るのは、失敗したときにこそ動いてほしい通知タスクです。UI のラベルだけでなく、API と YAML の `run_if` に書く大文字の値も一緒に覚えます。

境目は1本です。成功を要求する条件が満たされないと Upstream failed、失敗を要求する条件が満たされないと Excluded になります。前者はジョブ全体も失敗として扱われ、後者はジョブの成否に影響しません。上流が失敗したときだけ通知したいタスクに At least one failed を設定すれば、平常時は Excluded として静かに除外され、ジョブは成功のまま終わります。All done は依存の結果をいっさい問わないので、後片付けや一時ファイルの削除に向きます。割合で判定する条件は存在しません。

None failed と All succeeded は紛らわしいので分けておきます。None failed は「失敗した依存がゼロで、かつ少なくとも1つは実行された」で、スキップされた依存が混ざっていても成立します。All succeeded は全依存の成功を求めるため、こちらのほうが厳しい条件です。1つのタスクで後片付けと通知を兼ねることはできないので、役割ごとにタスクを分けます。

Excluded は成功として下流に伝わる

Excluded は失敗ではなく、条件が合わずに除外された状態です。この区別が下流の判定を左右します。

• Excluded の上流は、下流の Run if を評価するとき成功として扱われます。
• Upstream failed と Upstream canceled の上流は、スキップではなく失敗として扱われます。
• 依存がすべて Excluded になったタスクは、Run if の設定にかかわらず自分も Excluded になります。除外は依存の鎖に沿って下流へ連鎖します。

失敗時の通知ブランチを長く伸ばしたときに末尾のタスクが動かない理由は、たいてい3つめです。タスクを一時的に走らせたくないときは、削除ではなく無効化を使います。無効化すると設定と実行履歴を残したまま実行だけがスキップされ、下流タスクは自分の Run if 条件に照らして実行可否が判定されます。

If/else 条件タスクで値を見て分岐する

Run if が見ているのは上流の成否だけです。「行数が1,000以上なら全件ロード、そうでなければ差分ロード」のように値で分岐したいときは If/else 条件タスクを使います。ノートブックの中に `if` 文を埋め込むと、どちらに分岐したのかが実行画面に出ません。

条件は左オペランド・演算子・右オペランドの3要素です。`condition_task` には `left`、`op`、`right` を指定します。`op` に入る値は `EQUAL_TO`、`NOT_EQUAL`、`GREATER_THAN`、`GREATER_THAN_OR_EQUAL`、`LESS_THAN`、`LESS_THAN_OR_EQUAL` の6つです。オペランドにはジョブパラメータ、タスクパラメータ、タスク値の動的値参照を置けます。SQL 文や Python の式は書けません。

tasks:
  - task_key: check_rows
    condition_task:
      left: "{{tasks.count_rows.values.n}}"
      op: GREATER_THAN_OR_EQUAL
      right: "1000"
  - task_key: load_full
    depends_on:
      - task_key: check_rows
        outcome: "true"

分岐先を決めるのは `depends_on` の `outcome` です。指定できる値は `"true"` か `"false"` で、上流が If/else 条件タスクのときだけ意味を持ちます。UI では下流タスクの Depends on に `check_rows (true)` と表示されます。上流の成否を見る `run_if` とは役割が違います。

`==` と `!=` は文字列比較、`>`、`>=`、`<`、`<=` は数値比較です。したがって `12.0 == 12` は false になり、`12.0 >= 12` は true になります。しきい値の判定に等価比較を使うと、表記の揺れで静かに false になり、後続が丸ごと走らなくなります。ブール値は `"true"` と `"false"` という文字列に直されてから比較されます。

For each で同じ処理を回す

リージョンごとに同じ処理を回したいとき、タスクを対象の数だけコピーすると定義がすぐ壊れます。For each タスクは `inputs` に配列を渡し、`task` に定義したネストタスクを要素数だけ繰り返します。`inputs` の与え方は3通りで、それぞれ上限が違います。

ネストタスクの中では `{{input}}` が現在の反復の値全体を、`{{input.region}}` がオブジェクトの特定フィールドを指します。制約は3つです。第一に、Concurrency の既定は1で、指定しない限り反復は1つずつ順番に走り、所要時間は反復数に比例して伸びます。第二に、ネストタスクに別の For each タスクを置くことはできません。第三に、ネストタスクは下流から依存先にできないため、全反復の完了を待ちたい集計タスクは For each タスク本体に依存させます。ネストタスクが Pipeline タスクの場合は、Concurrency をいくつに設定しても1反復ずつになります。

分岐と反復の材料になるタスク値

分岐やループの条件は上流の計算結果から決まります。上流のノートブックが `dbutils.jobs.taskValues.set(key="n", value=1500)` で値を置き、下流はパラメータ欄に `{{tasks.count_rows.values.n}}` を書いて受け取ります。`set` と `get` は Python 関数で、言語に Python を選んだノートブックの中でしか呼べません。置かれた値のほうは、パラメータをサポートする他のタスク種別からも動的値参照で読めます。渡せるのは JSON として表現できる値で、その JSON 表現のサイズは48KiBを超えられません。DataFrame をそのまま渡すことはできません。

下流のノートブックで直接読む書き方は `dbutils.jobs.taskValues.get(taskKey="count_rows", key="n", debugValue=0)` です。ただし Databricks が推奨するのは、下流タスクのパラメータ欄に動的値参照を書く方法です。タスク名が変わったときに直す場所が1か所で済みます。

ここまでが構成の話です。組んだジョブをどのトリガーで動かし、失敗した実行を Repair run でどう救うかは次章で扱います。UI と Jobs API の機能網羅は記事「Lakeflow Jobs とは？ワークフロー管理完全ガイド」、宣言型パイプラインとの使い分けは記事「DLT と Workflows の違い」、コンピュートの選び方は記事「クラスタの種類」に譲ります。