ジョブを動かし、失敗から立て直す — トリガー・リトライ・パラメータ

11.1いつ動かすか — 5つのトリガーと Quartz cron / File arrival / Table update の制約値

「毎朝 2 時に動くはずの ETL が、気づくと 3 時に動いていた」「パートナーからのファイル到着が不定期なので、5 分おきの cron で一日中空振りさせている」。ジョブの起動まわりで起きる事故は、そのほとんどがトリガーの選び方と、選んだトリガーが持つ制約値を知らないことから生まれます。この節では 5 種類のトリガーを、要件のかたちから選び分けられるところまで整理します。

時刻が要件か、データが要件か

Lakeflow Jobs のトリガーは `Scheduled` (時刻ベース)、`File arrival` (ファイル到着)、`Table update` (テーブル更新)、`Continuous` (常時稼働)、`Manual (None)` (手動・外部起動) の 5 種類です。これに加えて、Unity Catalog のモデル更新で起動する Model update がベータ機能として用意されています。

選び分けの基準は単純です。実行時刻そのものが要件なら Scheduled、上流のデータ到着に追従したいなら File arrival か Table update を選びます。「毎営業日 07:00 に経営レポートを更新する」はデータが来ていなくても走らせる必要があるので Scheduled、「別チームが管理する Silver テーブルの更新が終わり次第 Gold 集計を走らせる」は Table update です。「データ駆動のほうが安いから常にそちらを選ぶ」という決め方はしません。起動タイミングの意味そのものが違います。

Scheduled — Simple と Advanced、そして Quartz cron の 6 フィールド

スケジュールの入力方法は Simple と Advanced の 2 つです。Simple は間隔と単位を入れるだけの簡易設定で、初回実行の時刻を指定できず、スケジューラが決めます。開始時刻とタイムゾーンを自分で決めたいなら Advanced を選び、「Show cron syntax」にチェックを入れると Quartz Cron 構文で直接表示・編集できます。

ここで最初につまずくのが cron の書式です。Databricks が採用しているのは Quartz 形式で、フィールドは `秒 分 時 日 月 曜日 [年]` の 6 個 (年は省略可) です。Linux の crontab で見慣れた 5 フィールドの `0 2 * * *` をそのまま貼ると、先頭が秒として解釈されて意図とまったく違う時刻になります。毎日 02:00:00 なら `0 0 2 * * ?`、平日 09:00 なら `0 0 9 ? * MON-FRI` と書きます。Quartz では「日」と「曜日」の両方を同時に指定できないため、片方に `?` (指定なし) を置くのが約束事です。

タイムゾーンも同じくらい事故が多い箇所です。指定しなければ UTC 扱いになるため、JST 02:00 のつもりが日本時間の 11:00 に走ります。一方で夏時間を採用するタイムゾーンを選ぶと、切り替えの日に毎時ジョブがスキップされたり、1〜2 時間遅れて見えたりします。絶対時刻で必ず動かしたいジョブは UTC を選ぶのが公式の推奨です。また、cron 式にどれだけ短い間隔を書いても、スケジュールによる連続実行の間隔は最小 10 秒に制限されます。1 秒間隔で回すことはできません。

File arrival — チェックは 1 分ごと、パスにワイルドカードは書けない

File arrival トリガーは、監視場所に新しいファイルが置かれたらジョブを起動する仕組みです。監視できるのは Unity Catalog の外部ロケーションまたはボリュームで、そのルートでもサブパスでも指定でき、配下は再帰的に見られます。DBFS のルート、ワークスペースのフォルダ、ドライバーノードのローカルパスは対象外です。

知らないと事故になる制約が 4 つあります。第一に、検知は約 1 分ごとのベストエフォートで、クラウドストレージの性能次第ではさらに遅れます。ミリ秒単位の即時起動を前提にした設計はできません。第二に、パスに `*` や `?` といったワイルドカードは書けず、外部テーブルやマネージドロケーションを含むパスも指定できません。第三に、起動するのは新規ファイルの到着だけで、同名ファイルの上書きでは起動しません。再処理させたければ別名で書き出す必要があります。第四に、ファイルイベントを有効にしていないストレージロケーションでは、File arrival トリガーを設定できるジョブはワークスペースあたり 50 個まで、監視場所に置けるファイル数は 10,000 個までという上限があります。設定にはストレージロケーションへの `READ` と、ジョブへの `CAN MANAGE` が必要です。

起動頻度を抑える設定は 2 つ用意されています。`Minimum time between triggers` は前の実行が完了してから次の起動までに空けるクールダウン、`Wait after last change` は最後の変更からこの秒数だけ新しいファイルが来なければ実行を開始するデバウンスで、ファイルが届くたびにタイマーがリセットされます。数十個のファイルがまとめて届く相手に対して 1 回だけ起動させたいなら後者です。なお、トリガーが担うのはジョブの起動までで、ファイルの読み取りとスキーマ処理は Auto Loader などタスク側の取り込みロジックの仕事です。両者は置き換え関係ではありません。

Table update — 1 トリガー 10 テーブル、権限は SELECT

Table update トリガーは、監視対象のテーブルが更新された時点でジョブを起動します。対象にできるのは Unity Catalog のマネージド Delta / Iceberg テーブル、Delta Lake に基づく外部テーブル、マテリアライズドビュー (クエリ結果を保存しておくテーブル)、ストリーミングテーブル (追記されたデータを増分で処理するテーブル)、および対応ソースに依存する Unity Catalog のビューやメトリックビューです。Hive metastore のテーブルは対象になりません。

上限は1 トリガーにつき最大 10 テーブルです。ビューを指定した場合はそのビューが依存するソーステーブルが上限にカウントされるため、11 テーブルに依存するビューはトリガーに使えません。ファイルイベントのない場所にあるテーブルの場合、Table update トリガーを設定できるジョブはワークスペースあたり 1,000 個までです。構成に必要な権限は監視対象テーブルへの `SELECT` だけで、メタストア管理者やカタログの所有者である必要はありません。

複数テーブルを登録したときは「Any table is updated」(どれか 1 つの更新で起動) と「All tables are updated」(すべてが更新されたら起動) を選びます。3 つの上流が揃ってから 1 回だけ走らせたいなら後者です。待機設定は File arrival と同じ 2 つが使えますが、併用したときの評価順は覚えておく価値があります。まず Minimum time between triggers を待ち、そのうえで最後の変更から Wait after last change が経過するのを待つという直列の判定です。120 秒と 60 秒を設定し、5 秒時点と 115 秒時点に更新があったなら、起動は 115 + 60 = 175 秒時点になります。設定画面の「Test trigger」ボタンを押せば、保存前に構成が有効かどうかを検証できます。

Continuous と Manual (None)

Continuous トリガーは常時稼働のストリーミング向けで、実行インスタンスは常に 1 つに制限され、実行が完了または失敗するたびに次の実行が自動的に始まります。実行が終わってから次が始まるまでの遅延は 60 秒未満です。失敗は指数バックオフで管理され、間隔を延ばしながらジョブ全体が自動的に再試行されます。ここで注意したいのは、Continuous ジョブでは通常のリトライポリシーを使えないことです。タスク単位でも再試行させたいなら Task retry mode に `On failure` を選び、ジョブ単位の再試行だけでよければ `Never` を選びます。止めたいときは Pause、再開は Resume です。1 分ごとの cron で常時稼働を模倣する必要はありません。

Manual (None) はトリガーを設定しない状態で、UI の実行ボタンか、外部のオーケストレーターからのプログラム的な呼び出しでのみ動きます。社内の Airflow などから起動し Databricks 側では自動起動させたくないなら、Scheduled を作って一時停止しておくのではなく、最初から Manual のままにします。

起動したあとの詰まり方 — 同時実行数とキュー

トリガーが引かれても、実行がすぐ始まるとは限りません。ここに関わるジョブレベルの既定値が 2 つあります。1 つ目は Maximum concurrent runs で、既定値は 1 です。前の実行がまだ走っているあいだに次のトリガーが来ても、同じジョブの実行は重なりません。日次バッチが 24 時間を超えたときに二重起動でテーブルを壊さないための安全弁ですが、パラメータ違いで同じジョブを並行実行したい場合はこの値を上げます。

2 つ目はキューイングで、既定で有効です。同時実行数の上限やワークスペースの上限に当たった実行は、その場で失敗するのではなくキューに入り、最大 48 時間までリソースの空きを待ちます。キューイングを有効にしても同時実行数の上限そのものが撤廃されるわけではない点に注意します。「トリガーは引かれたはずなのにクラスタが立っていない」という状況では、失敗を探す前にキューで待っていないかを確認します。

11.2止まったジョブを直す — リトライとタイムアウト、Repair run、通知4イベント

「外部 API が一瞬エラーを返しただけで夜間バッチが落ち、朝に見たら 8 タスク全部が未実行のままだった」。これは運用担当が寝ていたせいではなく、自動回復と部分再実行を設計していないことの結果です。この節では、勝手に立ち直らせる仕組みと、それでも落ちたときに人が最小の手数で立て直す手順を扱います。

リトライは 3 つのフィールド、タイムアウトは各試行ごとに効く

リトライはタスク単位の設定です。フィールドは `max_retries` (最大再試行回数)、`min_retry_interval_millis` (再試行の最小間隔をミリ秒で指定)、`retry_on_timeout` (タイムアウトしたときも再試行するか) の 3 つで、UI では Retries / Retry interval / Retry on timeout として出てきます。`retry_count` や `backoff_factor` といった名前のフィールドは存在しません。

この設定は継承されません。`depends_on` が制御するのは実行順序だけで、リトライ回数もパラメータも下流には伝播しません。バンドルの YAML では次のように書きます。

tasks:
  - task_key: ingest_api
    max_retries: 3
    min_retry_interval_millis: 300000
    retry_on_timeout: true
    timeout_seconds: 1800
  - task_key: publish
    max_retries: 0
    depends_on:
      - task_key: ingest_api

この定義では、外部 API を叩く ingest_api だけが 5 分間隔で最大 3 回まで自動再実行され、30 分でタイムアウトしたときも再試行の対象になります。publish は 1 回失敗したら確定します。

混同しやすいのが Timeout との関係です。Timeout を 30 分、Retries を 2 回に設定した場合、タイムアウトは初回とリトライの合計ではなく、各試行に個別に適用されます。30 分を超えた試行は Timed Out として打ち切られ、リトライが残っていれば改めて 30 分の枠で走り直します。最悪ケースの所要時間は 30 分ではなく 90 分です。「何時までに終わらせる」という約束から逆算するときは、足し算ではなく掛け算で考えます。

Timeout の隣に、所要時間しきい値の Warning という別の設定があります。こちらは想定時間を超えた時点でイベントを発生させて通知するだけで、処理そのものは継続します。「長引いていることは知りたいが、途中で止められるほうが困る」という要件は Timeout ではなく Warning です。逆に暴走を確実に止めたいときだけ Timeout を使います。

回数の決め方にも定石があります。リトライが効くのは、時間を置けば消える一時障害だけです。クラウドストレージのスロットリング、外部 API のレート制限、スポットインスタンス (安い代わりにクラウド側から回収されることがある計算資源) の回収などがこれにあたり、3 回・5 分間隔といった有限の設定で十分に救えます。一方で回数を 100 や無制限に引き上げるのは、恒久的な失敗を隠しながらコストだけを積み上げる典型的な悪手です。リトライを使い切ってもなお失敗したら通知を上げて人に渡す、という二段構えにします。

Repair run — 直すのは失敗タスクとその下流だけ

10 タスクのうち 7 番目が落ちたとき、Run now でジョブ全体を流し直すと成功済みの 6 タスクまで再計算されます。Repair run は、失敗・キャンセル・未実行のタスクと、それに依存する下流タスクだけを再実行する機能です。成功済みタスクの結果はそのまま使われます。ここで押さえるのは「失敗タスクだけ」ではなく「失敗タスクとその下流」という点で、上流の出力が作り直された以上、下流も計算し直さなければ整合が取れないからです。修復は元の run に紐付いた形で履歴が残り、何回修復したかは `{{job.repair_count}}` で参照できます。

使う前に知っておく制約が 3 つあります。まず、Repair run は 2 つ以上のタスクを持つジョブでのみ使えます。単一タスクのジョブが失敗した場合は Run now で流し直すしかありません。次に、修復時は現在のジョブ設定とタスク設定が使われ、修復ダイアログで入力したパラメータが既存の値を上書きします。最後に、共有のジョブクラスタを使っている場合、修復実行では `my_job_cluster_v1` のようなサフィックス付きの新しいジョブクラスタが作られます。

Run now は新しい run を作って先頭から全タスクを流し直す操作で、成功済みの重い ETL まで再計算し、履歴も別の run に分かれます。Repair run は同じ run の中に修復の反復として記録されるため、監査やコスト按分で「1 回の実行で何が起きたか」を一続きで追えます。

Repair run は冪等性を保証しない

ここが実務でいちばん刺さる論点です。公式ドキュメントは「Lakeflow Jobs はタスクを冪等にしない。タスクが失敗する前に出力の一部を書き込んでいた場合、再実行するとそのデータが重複しうる」と明記しています。Repair run は失敗したタスクを最初からやり直すだけで、途中まで書いたものを取り消してはくれません。リトライも同じで、3 回再試行するということは、半端な書き込みが 3 回積み上がる可能性があるということです。

したがって、リトライや Repair run を前提にするなら、タスク側を「何度走らせても結果が同じ」形に作る必要があります。よく使う型は 3 つです。1 つ目は追記の `INSERT INTO` をやめ、キーを指定した `MERGE INTO` で既存行を更新する形にすること。2 つ目は処理単位が日次なら `INSERT OVERWRITE` で対象範囲ごと置き換えること。3 つ目は書き込み先をいったん一時テーブルにして、最後の 1 コミットで本番テーブルに差し替えることです。いずれも「途中で落ちても中間状態が残らない」ことを狙った設計です。

直した結果を後から確かめる場所が run 履歴です。ここには開始・終了時刻、所要時間、ステータス、トリガー種別、パラメータ、タスクごとの出力とログが残り、UI と `GET /api/2.2/jobs/runs/list` から参照できます。保持期間はジョブ・パイプラインとも 60 日で、過ぎた run は自動的に消えるため、それ以上残したい記録は期限が来る前にエクスポートします。

通知は 4 イベントが基本

失敗に気づく仕組みが通知です。ジョブ単位でもタスク単位でも設定でき、購読できるイベントは Start (開始)、Success (成功)、Failure (失敗)、Duration warning (所要時間しきい値の超過) の 4 つが基本で、これに Streaming backlog (バックログがしきい値を超過) が加わります。宛先はメールのほか Slack、Microsoft Teams、PagerDuty、HTTP Webhook が使え、1 つのジョブまたはタスクにつき、イベント種別ごとに最大 3 つのシステム宛先を登録できます。ジョブ本体の処理には介入しないので、通知を足しても実行時間も DBU (Databricks の課金単位) も増えません。

現場で多い障害は「設定したのに届かない」です。原因の筆頭は On success だけにチェックが入っていて On failure が未購読というもので、SMTP やネットワークを疑う前にここを見ます。もう 1 つ引っかかりやすいのが、ジョブレベルの通知は、失敗したタスクがリトライされている間は送られないという挙動です。リトライ中の失敗も拾いたいならタスクレベルの通知を設定します。逆にノイズを減らしたいときは「スキップされた実行をミュート」「キャンセルされた実行をミュート」「最後のリトライまでミュート」を組み合わせます。なお、Webhook の配線や Slack 側の受け取り方といった通知チャネルの実装はこの章の範囲を超えるため、専用の記事に譲ります。

11.3値を渡す・外から動かす — パラメータ記法体系、taskValues 48 KiB、Jobs REST API と数値上限

「対象日をハードコードしたノートブックを、毎朝手で書き換えてから Run now を押している」。パラメータの渡し方を覚えていないと、ジョブが自動起動していても運用は自動化されません。この節では、値をどこからどう渡すかの記法体系、タスク間で値を受け渡す taskValues、そして外部から動かすための REST API と、本番設計で効いてくる数値上限をまとめます。

動的値参照は 4 つの系統で覚える

タスクの設定欄に `{{ }}` で書く記法を動的値参照と呼びます。種類が多く見えますが、値の出どころで 4 系統に整理できます。

• ジョブパラメータ: `{{job.parameters.<name>}}`。ジョブレベルで定義したキーと値を、どのタスクからでも参照します。
• タスク値: `{{tasks.<task_name>.values.<key>}}`。上流タスクが実行時に書き込んだ値を受け取ります。
• 反復値: `{{input}}` と `{{input.<key>}}`。For each タスクのネストされたタスクの中でだけ使えます。
• 実行メタデータ: `{{job.id}}`、`{{job.run_id}}`、`{{job.repair_count}}`、`{{task.execution_count}}` (リトライを含む実行回数)、`{{tasks.<task_name>.result_state}}` など。

日付を渡すときに使うのは `{{job.start_time.<argument>}}` です。引数には `iso_date`、`iso_datetime`、`year`、`month`、`day`、`hour`、`minute`、`second`、`timestamp_ms` が指定でき、パーティション値として `2026-08-09` のような文字列がほしいなら `{{job.start_time.iso_date}}` を書きます。`${run_date}` や `{{params.run_date}}` といった記法は存在しません。`{{job.trigger.type}}` を見れば何で起動されたかも分かり、値は `periodic` / `one_time` / `run_job_task` / `file_arrival` / `continuous` / `table` / `model` のいずれかです。

SQL タスクの結果を下流に渡す記法だけは系統が違います。`{{tasks.<task_name>.output.first_row.<column_alias>}}` でクエリ結果の先頭行から特定の列を参照でき、出力全体には 1,000 行かつ 48 KB という上限があります。件数を後段の分岐に使うときはこの記法です。

どこで定義し、どう受け取るか

記法が混乱するのは、定義する場所と受け取る場所を分けて覚えていないからです。次の表で 1 望できるようにします。

ノートブックタスクは、こうして渡ってきた値を `dbutils.widgets` で受け取ります。あらかじめウィジェットを定義しておき、`dbutils.widgets.get("env")` で読み出す形です。ウィジェットには `text`、`dropdown`、`combobox`、`multiselect` の 4 種類があり、UI からもジョブからも同じキーで値を注入できます。`sys.argv` で受けようとする実装は動きません。実行時に値を差し替えたいときは UI の「Run now with different parameters」を使います。

taskValues は Python 専用、JSON で 48 KiB まで

上流タスクの計算結果を下流に渡す公式の仕組みがタスク値です。上流で `dbutils.jobs.taskValues.set(key="record_count", value=1500)` と書き、下流タスクのパラメータ欄に `{{tasks.count_rows.values.record_count}}` を設定します。受け取り側で `dbutils.jobs.taskValues.get(taskKey="count_rows", key="record_count", debugValue=0)` と書くこともできますが、タスク名の変更に強いという理由で Databricks が推奨するのは動的値参照のほうです。`debugValue` はノートブックを対話的に試すときの代替値です。

制約は 2 つ覚えます。1 つは値の JSON 表現が 48 KiB を超えてはならないことです。JSON にできる値なら数値でもオブジェクトでも渡せますが、DataFrame をそのまま渡すことはできません。大きいデータはテーブルやボリュームに書き、パスやテーブル名だけをタスク値で渡します。もう 1 つは、`set` と `get` が Python 関数であるため、言語に Python を選んだノートブックの中でしか呼べないことです。SQL タスクの中から直接呼ぶことはできません。ただし設定された値そのものは、パラメータをサポートする他のタスク種別からも動的値参照で読み取れます。

渡した値の使われ方で 1 つだけ落とし穴があります。前章で扱った If/else 条件タスクは、`==` と `!=` をオペランドの文字列比較として評価するため `12.0 == 12` は false になり、`>` `>=` `<` `<=` は数値比較なので `12.0 >= 12` は true になります。件数のしきい値判定には大小比較を使います。For each タスクの Inputs も渡し方ごとに上限が違い、UI に JSON 配列を直接書く場合は 5,000 文字、タスク値を参照する場合は 48 KB、ジョブパラメータを参照する場合は 10,000 文字までです。

外から動かす — Jobs REST API 2.2

既存のジョブをパラメータ付きで即時実行するなら `POST /api/2.2/jobs/run-now` です。ボディには `job_id` と、ジョブレベルの `job_parameters` を渡します。あわせて `notebook_params`、`python_params`、`jar_params`、`sql_params` といったタスク種別ごとのパラメータも用意されており、ノートブックタスクだけに値を渡す古い形の `notebook_params` も現役です。ジョブ定義を持たない使い捨ての実行は `POST /api/2.2/jobs/runs/submit`、状態のポーリングは `GET /api/2.2/jobs/runs/get`、修復は `POST /api/2.2/jobs/runs/repair` です。`jobs/list` は一覧を返すだけで副作用はなく、`runs/delete` は履歴の削除であって起動ではありません。

実行結果を取りに行くのは `GET /api/2.2/jobs/runs/get-output` で、`dbutils.notebook.exit()` が返した値やエラー内容を取得できます。返るのは出力の先頭 5 MB までで、マルチタスクジョブでは親の run ではなく個々のタスクの run_id を指定します。

2.2 で変わった点も押さえます。`tasks` や `job_clusters` のようなリスト項目は 1 レスポンスあたり 100 要素までで、超えると `next_page_token` が返るので `page_token` に渡して続きを取ります。一覧系にあった `has_more` は廃止され、`next_page_token` の有無で判断します。`only_latest=true` を付けると、リトライや修復で置き換えられた古い試行を除いた最新の試行だけが返ります。キューイングは 2.2 で作ったジョブでは既定で有効です。実行履歴をまとめて集計したいときは API を回すのではなく、`system.lakeflow` スキーマのテーブルを SQL でクエリします。

暗記しておく数値上限

とくに 48 KiB と 30 MB / 8 MB は、超えたときのエラーメッセージがそっけないため、値を覚えていないと原因にたどり着くまでに時間を溶かします。API を使った監視やジョブ管理の自動化そのものは別の記事に譲ります。