はじめに
Lakeflow Jobsに、ジョブ完了トリガーがベータで追加されました。ほかのジョブが終わったことをきっかけに、自分のジョブを自動で起動できる機能です。
これまでも、ジョブ実行 (Run Job) タスクを使えば、あるジョブから別のジョブを呼び出せました。ただしその場合は、呼び出す側のジョブにタスクを追加する必要があります。ジョブ完了トリガーでは向きが逆になります。待つ側のジョブに「どのジョブの完了を待つか」を設定するので、上流のジョブには手を入れません。
他のチームが管理しているジョブに後続の処理をつなげたいとき、相手のジョブを変更してもらわずに済む。この点がこの機能の一番の使いどころだと思います。
今回は、上流のジョブ2本が両方とも成功したら下流のジョブが起動する構成を作りました。待ち状態がどう見えるか、下流のジョブが上流の情報をどう受け取れるかも含めて書いていきます。
ジョブ完了トリガーの動き
ドキュメントに書かれている動きのうち、今回の検証に関係するものを整理しておきます。
上流のジョブは、1つのトリガーにつき1〜10個まで指定できます。同じUnity Catalogメタストアにつながっていれば、別のワークスペースのジョブも指定できます。どの結果を起動のきっかけにするかは、次の3つから選びます。
| 上流のラン結果 | API の値 | 起動のきっかけになる完了 |
|---|---|---|
| 成功 (デフォルト) | SUCCESS |
SUCCESS と SUCCESS_WITH_FAILURES
|
| 不成功 | FAILED |
SUCCESS 以外で終わったランと SUCCESS_WITH_FAILURES
|
| 結果を問わない | DONE |
終わったすべてのラン |
上流のジョブが複数ある場合は、すべてのジョブに新しい完了がそろった時点で下流のジョブが起動します。どれか1つが終わったら起動する、という動きにはなりません。そうしたい場合は、上流のジョブごとに別のトリガーを作ります。これには複数トリガーのベータも有効にしておく必要があります。
注意しておきたいのは、完了を営業日やデータのバッチ単位で突き合わせないことです。「同じ日付のデータを処理したAとBがそろったら起動する」という意味の待ち合わせではありません。各ジョブの最新の完了がそろったかどうかだけを見ます。
検証の構成
上流のジョブ2本と下流のジョブ1本を用意しました。
| ジョブ | 名前 | やること |
|---|---|---|
| A | jct-demo-upstream-a |
10秒待って終わる |
| B | jct-demo-upstream-b |
10秒待って終わる |
| C | jct-demo-downstream-c |
上流の完了情報を受け取って表示する |
ジョブCに「AとBが成功したら起動する」トリガーを設定し、次の順に確かめます。
- Aだけを実行し、Cが起動しないこと
- Bを実行し、Cが起動すること
- Cが受け取った上流の完了情報に、AとBのランが入っていること
ジョブとタスク用のノートブックはDatabricks SDKで作り、トリガーの設定はGUIで行いました。ジョブはすべてサーバーレスで実行しています。
準備
ベータを有効にする
ワークスペースのプレビューページで「Job Completion Triggers」をオンにします。
ドキュメントによると、下流のジョブがあるワークスペースと、上流のジョブがあるすべてのワークスペースで有効にする必要があります。今回は1つのワークスペースで完結させたので、1か所だけで済みました。
下流のジョブで動かすノートブック
ジョブCで動かすノートブックです。上流の完了情報は、ジョブパラメーター upstream_completions として受け取ります。パラメーターの設定はあとでGUIから行います。
import json
# ジョブパラメーター upstream_completions で上流の完了情報を受け取る
# 手動で実行したときは空の配列 [] になる
dbutils.widgets.text("upstream_completions", "[]")
raw = dbutils.widgets.get("upstream_completions")
print(raw)
completions = json.loads(raw)
for c in completions:
print(f"job_id={c['job_id']} run_id={c['run_id']} "
f"termination_code={c['termination_code']} completion_time={c['completion_time']}")
# 呼び出し側から値を確認できるよう、受け取った値をそのまま出力として返す
dbutils.notebook.exit(raw)
ジョブA・Bで動かすノートブックは、10秒待って終わるだけです。
import time
time.sleep(10)
print("finished")
ジョブを作る
ノートブックをワークスペースに置いたうえで、SDKでジョブを作ります。クラスターを指定していないので、タスクはサーバーレスで実行されます。
from databricks.sdk import WorkspaceClient
from databricks.sdk.service import jobs as j
w = WorkspaceClient()
me = w.current_user.me().user_name
BASE_DIR = f"/Users/{me}/job_completion_trigger_demo"
jobs_to_create = {
"jct-demo-upstream-a": f"{BASE_DIR}/upstream_a",
"jct-demo-upstream-b": f"{BASE_DIR}/upstream_b",
"jct-demo-downstream-c": f"{BASE_DIR}/downstream_c",
}
for name, path in jobs_to_create.items():
created = w.jobs.create(
name=name,
tasks=[j.Task(task_key="main", notebook_task=j.NotebookTask(notebook_path=path))],
)
print(name, created.job_id)
これでジョブが3本できます。上流のジョブA・Bの設定は、このあと一切変更しません。
下流のジョブにトリガーを設定する
ここからはGUIで、ジョブCだけを操作します。
トリガーのタイプを選ぶ
ジョブCを開き、右側の「スケジュールとトリガー」で「トリガーを追加」をクリックします。トリガータイプで「アップストリームジョブの完了」を選ぶと、上流のジョブを指定する欄が出てきます。
ワークスペースを選び、ジョブ名かジョブIDで上流のジョブを検索します。ワークスペースの欄では、今いるワークスペースが選ばれた状態になっています。一覧に出るのは、自分に閲覧権限があるジョブだけです。
上流のジョブを2つ指定する
「アップストリームジョブを追加」で、ジョブAとジョブBを指定します。上流のラン結果は「成功」のままにしました。
上流のジョブを2つにすると、画面上部の説明文が変わります。1つのときは「Run this job when the selected upstream job completes ...」でしたが、2つにすると「Run this job after all of the selected upstream jobs complete ...」になります。すべてのジョブがそろうのを待つ、という動きが画面からも読み取れます。
Test triggerで権限を確かめる
保存する前に「Test trigger」をクリックします。
「You have access to all monitored jobs」と表示されました。ドキュメントによると、トリガーが上流のジョブを監視できるかどうかは、下流のジョブの実行ユーザーの権限で決まります。実行ユーザーには、監視するすべてのジョブへの CAN VIEW 以上の権限が必要です。権限が足りないと上流のジョブを監視できないので、保存の前に一度押しておくのがよいと思います。
保存すると、「スケジュールとトリガー」に「アップストリームジョブの完了時 (2件のジョブ、成功)」と表示されます。
上流の完了情報をパラメーターで受け取る
同じ画面の「ジョブパラメーター」で「パラメーターを編集」をクリックし、次のパラメーターを追加します。
| キー | 値 |
|---|---|
upstream_completions |
{{job.trigger.job_completion.completed_jobs}} |
キーの名前は、ドキュメントの例に合わせました。このパラメーター参照を使えるのは、ジョブのトリガーがすべてジョブ完了トリガーである場合に限られます。スケジュールなど別の種類のトリガーと組み合わせていると使えないので、この点は覚えておくとよさそうです。
保存した設定をAPIで見る
GUIで保存した設定がAPIからどう見えるかも確認しておきます。ベータで追加された項目をSDKのクラスを通さずに辞書のまま受け取るため、REST APIを直接呼んでいます。
import json
# include_trigger_state=true を付けると、トリガーの待ち状態も返る
job = w.api_client.do(
"GET", "/api/2.2/jobs/get",
query={"job_id": JOB_C_ID, "include_trigger_state": "true"},
)
print(json.dumps(job["settings"]["trigger"], indent=2))
print(json.dumps(job["trigger_state"], indent=2))
トリガーの設定は次のように保存されていました。
{
"pause_status": "UNPAUSED",
"job_completion": {
"jobs": [
{
"workspace_id": "5099015744649857",
"job_id": "1120223026356170"
},
{
"workspace_id": "5099015744649857",
"job_id": "822026838442657"
}
],
"if_status": "SUCCESS"
}
}
上流のジョブは、ワークスペースIDとジョブIDの組で指定されています。GUIで選んだ「成功」は if_status: SUCCESS として保存されています。
trigger_state には、監視しているジョブごとの状態が入っています。まだどのジョブも実行していないので、どちらも has_seen_completion: false です。
{
"job_completion": {
"monitored_job_states": [
{
"workspace_id": "5099015744649857",
"job_id": "1120223026356170",
"has_seen_completion": false,
"is_deleted": false
},
{
"workspace_id": "5099015744649857",
"job_id": "822026838442657",
"has_seen_completion": false,
"is_deleted": false
}
]
}
}
Aだけが終わってもCは起動しない
ジョブAだけを実行しました。Aは11:37:51に成功で終わりましたが、そこから180秒待ってもジョブCのランは0件のままでした。
ジョブCの画面には、トリガーを評価した結果が表示されています。
「2026年11月04日 11:37に評価済み、実行はトリガーされませんでした」とあり、その下に理由として「Not all monitored jobs have completed.」と出ています。Aの完了をきっかけにトリガーが評価され、Bの完了がまだないので起動しなかった、ということが画面から分かります。この画面では、先ほど設定したジョブパラメーターも確認できます。
この時点の trigger_state を見ると、ジョブAの欄だけが更新されていました。
{
"job_completion": {
"monitored_job_states": [
{
"workspace_id": "5099015744649857",
"job_id": "1120223026356170",
"last_seen_completion_time": "2026-11-04T02:37:51.389Z",
"last_seen_run_id": "637382874547417",
"has_seen_completion": true,
"last_seen_result": "SUCCESS",
"is_deleted": false
},
{
"workspace_id": "5099015744649857",
"job_id": "822026838442657",
"has_seen_completion": false,
"is_deleted": false
}
]
}
}
Aは has_seen_completion: true になり、完了したランのIDと時刻、結果が記録されています。Bは false のままです。下流のジョブが起動しないときに、どの上流のジョブを待っているのかをこのJSONで確かめられます。運用で「なぜ動かないのか」を調べるときは、ここを見るのが早そうです。
Bが終わるとCが起動する
続けてジョブBを実行しました。Bは11:41:15に成功で終わり、その52秒後の11:42:07にジョブCが起動しました。この待ち時間は、今回の環境で1回測っただけの値です。
Aの実行からCの起動までを時刻の順に並べると、次のようになります。
ジョブCのランの一覧では、起動方法が「ジョブ完了をトリガーとして」になっています。
Runs APIでジョブCのランを取得すると、trigger が JOB_COMPLETION になっていました。trigger_info には、起動のきっかけになった上流のランが入っています。
{
"job_completion": {
"completed_jobs": [
{
"workspace_id": "5099015744649857",
"job_id": "1120223026356170",
"run_id": "637382874547417",
"termination_code": "SUCCESS",
"completion_time": "2026-11-04T02:37:51.389Z"
},
{
"workspace_id": "5099015744649857",
"job_id": "822026838442657",
"run_id": "447223468525424",
"termination_code": "SUCCESS",
"completion_time": "2026-11-04T02:41:15.853Z"
}
]
}
}
run_id は、実行したAとBのランとそれぞれ一致しています。
下流のジョブが受け取った値
ジョブCのノートブックが受け取った upstream_completions の値は次のとおりでした。
[{"workspace_id":"5099015744649857","job_id":"1120223026356170","run_id":"637382874547417","termination_code":"SUCCESS","completion_time":"2026-11-04T02:37:51.389Z"},{"workspace_id":"5099015744649857","job_id":"822026838442657","run_id":"447223468525424","termination_code":"SUCCESS","completion_time":"2026-11-04T02:41:15.853Z"}]
trigger_info と同じ内容が、上流のジョブごとに1要素のJSON配列として渡されています。completion_time は数値ではなく、UTCのISO 8601形式の文字列でした。日本時間で扱う場合は変換が必要です。
この値があれば、下流のノートブックで上流のランIDを記録したり、上流のランの出力を読みに行ったりできます。「どの上流の実行を受けて動いたのか」を下流の処理の中で持てるのは、あとから処理を追いかけるときに効いてきます。
ここで一つハマりどころがあります。Runs APIでジョブCのランを取得し、job_parameters を見ると、値はテンプレートの文字列のままでした。
[
{
"name": "upstream_completions",
"default": "{{job.trigger.job_completion.completed_jobs}}"
}
]
展開後の値は、ここには入っていません。ランの外から上流の完了情報を確かめたい場合は、trigger_info.job_completion.completed_jobs を読みます。
まとめ
ジョブ完了トリガーを試して分かったことをまとめます。
- 上流のジョブには何も設定せず、下流のジョブの設定だけで、複数ジョブの完了を待ってから起動する構成を組める
- 上流のジョブが複数ある場合は、すべてのジョブに新しい完了がそろってから起動する。Aだけの完了ではCは起動しなかった
- 「Test trigger」で、実行ユーザーが監視するジョブを閲覧できるかを保存前に確かめられる
- 起動しなかった理由は、ジョブの画面に「Not all monitored jobs have completed.」と表示される
include_trigger_state=trueでジョブを取得すると、上流のジョブごとの待ち状態 (has_seen_completion、last_seen_run_id) が分かる- 下流のジョブは、パラメーター
{{job.trigger.job_completion.completed_jobs}}で上流のランIDや完了時刻をJSON配列として受け取れる - ランの
job_parametersにはテンプレートの文字列がそのまま入る。ランの外から上流の情報を見るときはtrigger_infoを読む
一番の収穫は、待ち状態を画面とAPIの両方で確かめられると分かったことでした。完了待ちの仕組みは、動かないときに理由が見えないと運用で困ります。この機能では、どの上流のジョブを待っているのかを trigger_state で確かめられるので、上流が他チームのジョブでも原因を追いやすいと感じました。今回は同じワークスペースの中で試しましたが、次は別のワークスペースのジョブを上流にした構成も試してみたいと思います。