databricks-jobs

Lakeflow Jobs の開発

最初に：CLI の基本、認証、プロファイルの選択、データ探索コマンドについては、親の databricks-core スキルを使用してください。

コンピューティング：ジョブはデフォルトでサーバーレスで実行されます。コンピューティングを選択したり、クラスター設定を記述したりする前に、databricks-serverless スキルを使用してください。サーバーレスとクラシックのどちらを使うか、およびパフォーマンスモードが決まります。

Lakeflow Jobs は、複数のタスクからなる DAG、柔軟なトリガー、包括的な監視機能でデータワークフローをオーケストレーションします。Jobs はさまざまなタスクタイプをサポートしており、Declarative Automation Bundles（DABs）、Python SDK、CLI を使用して管理できます。

リファレンスファイル

ユースケース	リファレンスファイル
タスクタイプを設定する（notebook、Python、SQL、dbt、pipeline、JAR、run_job、for_each）	references/task-types.md
トリガーとスケジュールを設定する（cron、定期実行、ファイル到着、テーブル更新、継続実行）	references/triggers-schedules.md
通知、ヘルスルール、再試行、タイムアウト、キューを設定する	references/notifications-monitoring.md
一連の実例（ETL、ウェアハウスの更新、イベント駆動型、ML トレーニング、複数環境、ストリーミング、ジョブ間連携）	references/examples.md

新しいジョブプロジェクトのひな形作成

databricks bundle init と設定ファイルを使用して、対話なしでひな形を作成します。これにより、<project_name>/ ディレクトリにプロジェクトが作成されます。

databricks bundle init default-python --config-file <(echo '{"project_name": "my_job", "include_job": "yes", "include_pipeline": "no", "include_python": "yes", "serverless": "yes"}') --profile <PROFILE> < /dev/null

• project_name：使用できるのは英字、数字、アンダースコアのみです

bundle init はプロジェクト内に AGENTS.md も作成します。新しいエージェント指示ファイルを作成するのではなく、プロジェクトに合わせてそのファイルをカスタマイズしてください。

プロジェクト構成

my-job-project/
├── databricks.yml              # Bundle configuration
├── resources/
│   └── my_job.job.yml          # Job definition
├── src/
│   ├── my_notebook.ipynb       # Notebook tasks
│   └── my_module/              # Python wheel package
│       ├── __init__.py
│       └── main.py
├── tests/
│   └── test_main.py
└── pyproject.toml              # Python project config (if using wheels)

クイックスタート

Declarative Automation Bundles（DABs）— 推奨

# resources/jobs.yml
resources:
  jobs:
    my_etl_job:
      name: "[${bundle.target}] My ETL Job"
      tasks:
        - task_key: extract
          notebook_task:
            notebook_path: ../src/notebooks/extract.py
      # no cluster block = serverless; compute details: databricks-serverless

Python SDK

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.jobs import Task, NotebookTask, Source

w = WorkspaceClient()

job = w.jobs.create(
    name="my-etl-job",
    tasks=[
        Task(
            task_key="extract",
            notebook_task=NotebookTask(
                notebook_path="/Workspace/Shared/etl/extract",
                source=Source.WORKSPACE,
            ),
        ),
    ],
)
print(f"Created job: {job.job_id}")

CLI

databricks jobs create --json '{
  "name": "my-etl-job",
  "tasks": [{
    "task_key": "extract",
    "notebook_task": {
      "notebook_path": "/Workspace/Shared/etl/extract",
      "source": "WORKSPACE"
    }
  }]
}'

主な概念

複数タスクのワークフロー

Jobs は DAG ベースのタスク依存関係をサポートしています。

tasks:
  - task_key: extract
    notebook_task:
      notebook_path: ../src/extract.py

  - task_key: transform
    depends_on:
      - task_key: extract
    notebook_task:
      notebook_path: ../src/transform.py

  - task_key: load
    depends_on:
      - task_key: transform
    run_if: ALL_SUCCESS  # Only run if all dependencies succeed
    notebook_task:
      notebook_path: ../src/load.py

run_if 条件：

• ALL_SUCCESS（デフォルト）— すべての依存関係が成功した場合に実行します
• ALL_DONE — すべての依存関係が完了した場合に実行します（成功または失敗）
• AT_LEAST_ONE_SUCCESS — 少なくとも1つの依存関係が成功した場合に実行します
• NONE_FAILED — 失敗した依存関係がない場合に実行します
• ALL_FAILED — すべての依存関係が失敗した場合に実行します
• AT_LEAST_ONE_FAILED — 少なくとも1つの依存関係が失敗した場合に実行します

タスクタイプの概要

タスクタイプ	ユースケース	リファレンス
notebook_task	ノートブックを実行する	references/task-types.md#notebook-task
spark_python_task	Python スクリプトを実行する	references/task-types.md#spark-python-task
python_wheel_task	Python wheel を実行する	references/task-types.md#python-wheel-task
sql_task	SQL クエリ／ファイル／ダッシュボード／アラートを実行する	references/task-types.md#sql-task
dbt_task	dbt プロジェクトを実行する	references/task-types.md#dbt-task
pipeline_task	SDP（旧 DLT）パイプラインをトリガーする	references/task-types.md#pipeline-task
spark_jar_task	Spark JAR を実行する	references/task-types.md#spark-jar-task
run_job_task	他のジョブをトリガーする	references/task-types.md#run-job-task
for_each_task	入力を反復処理する	references/task-types.md#for-each-task

トリガータイプの概要

トリガータイプ	用途	参照
schedule	Cron ベースのスケジュール設定	references/triggers-schedules.md#cron-schedule
trigger.periodic	間隔ベース	references/triggers-schedules.md#periodic-trigger
trigger.file_arrival	ファイル到着イベント	references/triggers-schedules.md#file-arrival-trigger
trigger.table_update	Unity Catalog のテーブル変更イベント	references/triggers-schedules.md#table-update-trigger
continuous	常時実行されるジョブ	references/triggers-schedules.md#continuous-jobs

コンピュートの設定

デフォルトはサーバーレスです。クラスター設定（job_clusters、new_cluster、job_cluster_key、 existing_cluster_id）は省略してください。これらを指定しないノートブック、Python スクリプト、wheel、JAR タスクはサーバーレスで実行されます。 SQL タスクと dbt タスクの SQL は SQL ウェアハウスで実行されます（warehouse_id。詳細は タスクタイプを参照）。パイプラインタスクはパイプライン固有のコンピュートで実行され （パイプライン内で serverless: true を指定）、ジョブの performance_target でそのモードを設定します。 ワークスペースでサーバーレスが利用できないためにジョブの作成または実行に失敗する場合は、 ジョブ全体をクラシックに設定し、その旨を伝えてください（ジョブコンピュートで実行するタスクにはジョブクラスターを使用します）。

コンピュートに関するその他すべての設定では、databricks-serverless スキルを使用してください。対象となるのは、パフォーマンスモード （performance_target）、ライブラリ（environments）、サーバーレスで実行できない可能性のある既存コード、クラシッククラスターが必要となる一部のワークロードです。 このスキルがインストールされていない場合は、databricks aitools install を実行してください。いずれの場合も、すべてのジョブに performance_target を設定します。コストを最適化するには STANDARD を、ジョブの実行時間が重要な場合、対話型の場合、または他のワークロードがジョブに依存する場合は PERFORMANCE_OPTIMIZED を指定してください。

ジョブパラメーター

ジョブレベルで定義したパラメーターは、すべてのタスクに渡されます（タスクごとに繰り返し指定する必要はありません）。

parameters:
  - name: env
    default: "dev"
  - name: date
    default: "{{start_date}}"  # Dynamic value reference

ノートブックでのアクセス方法：

catalog = dbutils.widgets.get("env")
load_date = dbutils.widgets.get("date")

特定のタスクへの受け渡し方法：

tasks:
  - task_key: my_task
    notebook_task:
      notebook_path: ../src/notebook.py
      base_parameters:
        env: "{{job.parameters.env}}"
        custom_param: "value"

一般的な操作

Python SDK

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

# List jobs
jobs = w.jobs.list()

# Get job details
job = w.jobs.get(job_id=12345)

# Run job now
run = w.jobs.run_now(job_id=12345)

# Run with parameters
run = w.jobs.run_now(
    job_id=12345,
    job_parameters={"env": "prod", "date": "2024-01-15"},
)

# Cancel run
w.jobs.cancel_run(run_id=run.run_id)

# Delete job
w.jobs.delete(job_id=12345)

CLI

# List jobs
databricks jobs list

# Get job details
databricks jobs get 12345

# Run job
databricks jobs run-now 12345

# Run with parameters (must use --json with job_id inside)
databricks jobs run-now --json '{"job_id": 12345, "job_parameters": {"env": "prod"}}'

# Cancel run
databricks jobs cancel-run 67890

# Delete job
databricks jobs delete 12345

Declarative Automation Bundle の操作

# Validate configuration
databricks bundle validate --profile <profile>

# Deploy to a target
databricks bundle deploy -t dev --profile <profile>

# Run a job
databricks bundle run <job_name> -t dev --profile <profile>

# Check run status
databricks jobs get-run --run-id <id> --profile <profile>

# Destroy resources
databricks bundle destroy --auto-approve

権限（DAB）

resources:
  jobs:
    my_job:
      name: "My Job"
      permissions:
        - level: CAN_VIEW
          group_name: "data-analysts"
        - level: CAN_MANAGE_RUN
          group_name: "data-engineers"
        - level: CAN_MANAGE
          user_name: "[email protected]"

権限レベル：

• CAN_VIEW — ジョブと実行履歴の表示
• CAN_MANAGE_RUN — 実行の表示、開始、キャンセル
• CAN_MANAGE — 編集や削除を含むすべての操作

単体テスト

単体テストをローカルで実行します：

uv run pytest

開発ワークフロー

1. 検証：databricks bundle validate --profile <profile>
2. デプロイ：databricks bundle deploy -t dev --profile <profile>
3. 実行：databricks bundle run <job_name> -t dev --profile <profile>
4. 実行ステータスの確認：databricks jobs get-run --run-id <id> --profile <profile>

よくある問題

問題	解決策
サーバーレスジョブの起動が遅い	performance_target: PERFORMANCE_OPTIMIZED を指定してください。databricks-serverless を参照してください
タスクの依存関係が機能しない	depends_on 内の task_key 参照が完全に一致していることを確認してください
スケジュールがトリガーされない	pause_status: UNPAUSED と有効なタイムゾーンを確認してください
ファイル到着が検出されない	パスに適切な権限があり、クラウドストレージの URL を使用していることを確認してください
テーブル更新トリガーでイベントが欠落する	Unity Catalog のテーブルと適切な権限付与を確認してください
パラメーターにアクセスできない	ノートブックで dbutils.widgets.get() を使用してください
admins グループのエラー	ジョブの admins 権限は変更できません
サーバーレスタスクが失敗する	databricks-serverless の「よくある問題」を参照してください（ライブラリ、RDD コード、DBFS マウント）

関連スキル

• databricks-dabs — ジョブとパイプラインで共通の DAB 設定パターン
• databricks-pipelines — pipeline_task によってトリガーされる SDP（旧 DLT）パイプライン
• databricks-serverless — サーバーレス、パフォーマンスモード、クラシックへのフォールバックに関するコンピュートの選択

Documentation

• Lakeflow Jobs
• Task types
• Declarative Automation Bundles
• Jobs API Reference
• Bundle Examples Repository