databricks-jobs

Lakeflow Jobs 开发

首先：使用上级 databricks-core 技能了解 CLI 基础知识、身份验证、配置文件选择和数据探索命令。

计算资源：作业默认在无服务器环境中运行。选择计算资源或编写任何集群设置之前，请使用 databricks-serverless 技能；该技能会决定使用无服务器还是经典计算，以及性能模式。

Lakeflow Jobs 使用多任务 DAG、灵活的触发器和全面的监控来编排数据工作流。作业支持多种任务类型，可通过 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。请自定义该文件以适配项目， 不要新建 Agent 指令文件。

项目结构

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"
    }
  }]
}'

核心概念

多任务工作流

作业支持基于 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 — 至少一个依赖项成功时运行
• NONE_FAILED — 没有依赖项失败时运行
• ALL_FAILED — 所有依赖项均失败时运行
• AT_LEAST_ONE_FAILED — 至少一个依赖项失败时运行

任务类型摘要

任务类型	用例	参考
notebook_task	运行 notebook	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）pipeline	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）；未配置这些设置的 Notebook、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

在 Notebook 中访问：

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

声明式自动化 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 表和相应授权
无法访问参数	在 Notebook 中使用 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