Databricks Jobs APIで実現するジョブ自動化・CI/CD・監視

Databricksのジョブ管理はUIでの手動操作でも可能ですが、本番環境での運用やCI/CDパイプラインの構築には Jobs APIの理解が不可欠です。この記事では、Jobs API 2.1のエンドポイント構成、リクエスト例、 ジョブ定義のJSON構造、認証方式、そしてRuns APIによる実行監視までを体系的に整理します。

Jobs API 2.1 エンドポイント一覧

Jobs API 2.1で利用する主要エンドポイントを以下に整理します。 すべてのエンドポイントのベースURLは https://<workspace-url>/api/2.1/jobs/ です。

resetはジョブ定義全体を置き換える（指定しなかったフィールドはデフォルトに戻る）のに対し、updateは指定したフィールドだけを変更します。CI/CDでは定義全体をコードで管理するためresetが多用されます。

curlリクエスト例

以下はPATを使ってジョブを作成し、実行するcurl例です。

# ジョブ作成
curl -X POST \
  https://adb-1234567890.12.azuredatabricks.net/api/2.1/jobs/create \
  -H "Authorization: Bearer dapi_xxxxxxxxxxxx" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "daily_etl_pipeline",
    "tasks": [{
      "task_key": "ingest",
      "notebook_task": {
        "notebook_path": "/Repos/prod/etl/ingest"
      },
      "new_cluster": {
        "spark_version": "14.3.x-scala2.12",
        "node_type_id": "i3.xlarge",
        "num_workers": 4
      }
    }]
  }'

# ジョブ実行（run-now）
curl -X POST \
  https://adb-1234567890.12.azuredatabricks.net/api/2.1/jobs/run-now \
  -H "Authorization: Bearer dapi_xxxxxxxxxxxx" \
  -H "Content-Type: application/json" \
  -d '{"job_id": 12345}'

Pythonリクエスト例

requestsライブラリを使った同等の処理です。本番ではエラーハンドリングとリトライを追加します。

import requests

WORKSPACE_URL = "https://adb-1234567890.12.azuredatabricks.net"
TOKEN = "dapi_xxxxxxxxxxxx"
HEADERS = {"Authorization": f"Bearer {TOKEN}"}

# ジョブ作成
job_config = {
    "name": "daily_etl_pipeline",
    "tasks": [{
        "task_key": "ingest",
        "notebook_task": {"notebook_path": "/Repos/prod/etl/ingest"},
        "new_cluster": {
            "spark_version": "14.3.x-scala2.12",
            "node_type_id": "i3.xlarge",
            "num_workers": 4
        }
    }]
}
resp = requests.post(
    f"{WORKSPACE_URL}/api/2.1/jobs/create",
    headers=HEADERS, json=job_config
)
job_id = resp.json()["job_id"]

# 即時実行
run_resp = requests.post(
    f"{WORKSPACE_URL}/api/2.1/jobs/run-now",
    headers=HEADERS, json={"job_id": job_id}
)
run_id = run_resp.json()["run_id"]

# 実行結果のポーリング
import time
while True:
    status = requests.get(
        f"{WORKSPACE_URL}/api/2.1/jobs/runs/get",
        headers=HEADERS, params={"run_id": run_id}
    ).json()
    life_cycle = status["state"]["life_cycle_state"]
    if life_cycle in ("TERMINATED", "INTERNAL_ERROR", "SKIPPED"):
        break
    time.sleep(30)

ジョブ定義JSON構造（マルチタスク）

Jobs API 2.1の最大の特徴はマルチタスクジョブです。複数のタスクを1つのジョブ内に定義し、depends_onで依存関係を宣言的に記述できます。

{
  "name": "daily_pipeline",
  "schedule": {
    "quartz_cron_expression": "0 0 6 * * ?",
    "timezone_id": "Asia/Tokyo"
  },
  "tasks": [
    {
      "task_key": "extract",
      "notebook_task": {"notebook_path": "/Repos/prod/etl/extract"},
      "new_cluster": { ... }
    },
    {
      "task_key": "transform",
      "depends_on": [{"task_key": "extract"}],
      "notebook_task": {"notebook_path": "/Repos/prod/etl/transform"},
      "new_cluster": { ... }
    },
    {
      "task_key": "load",
      "depends_on": [{"task_key": "transform"}],
      "notebook_task": {"notebook_path": "/Repos/prod/etl/load"},
      "new_cluster": { ... }
    }
  ],
  "email_notifications": {
    "on_failure": ["[email protected]"]
  },
  "max_concurrent_runs": 1
}

max_concurrent_runsを1にすると、前回のRunが完了するまで次のRunは開始されません。 スケジュール起動とrun-now APIの競合を防ぐために、本番ジョブでは1に設定するのが一般的です。

パラメータ渡し

run-now APIでジョブを実行する際に、パラメータを動的に渡すことができます。 ノートブックタスクではbase_parameters（ジョブ定義側のデフォルト）とnotebook_params（run-now側の上書き）で値を受け渡します。

# run-nowでパラメータ上書き
curl -X POST .../api/2.1/jobs/run-now \
  -d '{
    "job_id": 12345,
    "notebook_params": {
      "target_date": "2026-03-27",
      "mode": "full_refresh"
    }
  }'

ノートブック内ではdbutils.widgets.get("target_date")でパラメータを取得します。 Pythonタスクではpython_params、SparkSubmitタスクではspark_submit_paramsを使います。

Runs API による実行監視

ジョブの実行結果を確認するにはRuns APIを使います。重要なフィールドは以下の通りです。

API認証方式の比較

試験で問われるポイント

• 「resetとupdateの違い」→ resetは定義全体の上書き、updateは部分変更
• 「run-nowとsubmitの違い」→ run-nowは既存ジョブの実行、submit（runs/submit）は一時的なジョブ定義で即時実行
• 「max_concurrent_runsの目的」→ 同一ジョブの同時実行数を制限し、リソース競合やデータ不整合を防止
• 「CI/CDパイプラインに適した認証方式」→ OAuth M2M（Service Principal）

問題で確認

よくある質問