将任务添加到 Databricks 资产捆绑包中的作业

本页提供有关如何在 Databricks 资产捆绑包中定义作业任务的信息。 有关任务作业的信息，请参阅 在 Lakeflow 中配置和编辑任务。

配置任务

在tasks键中为捆绑包中的作业定义任务。 可用任务类型的任务配置示例位于 “任务设置” 部分。 有关在捆绑包中定义作业的信息，请参阅 作业。

若要设置任务值，大多数作业任务类型都具有特定于任务的参数，但你也可以定义传递给任务的 作业参数 。 作业参数支持动态值引用，这些参数支持传递特定于任务之间作业运行的值。 有关如何按任务类型传递任务值的完整信息，请参阅 按任务类型的详细信息。

还可以使用目标工作区的设置替代常规作业任务设置。 请参阅 “使用目标设置进行替代”。

以下示例配置定义包含两个笔记本任务的作业，并将任务值从第一个任务传递到第二个任务。

resources:
  jobs:
    pass_task_values_job:
      name: pass_task_values_job
      tasks:
        # Output task
        - task_key: output_value
          notebook_task:
            notebook_path: ../src/output_notebook.ipynb

        # Input task
        - task_key: input_value
          depends_on:
            - task_key: output_value
          notebook_task:
            notebook_path: ../src/input_notebook.ipynb
            base_parameters:
              received_message: '{{tasks.output_value.values.message}}'

output_notebook.ipynb包含以下代码，用于设置message键的任务值：

# Databricks notebook source
# This first task sets a simple output value.

message = "Hello from the first task"

# Set the message to be used by other tasks
dbutils.jobs.taskValues.set(key="message", value=message)

print(f"Produced message: {message}")

input_notebook.ipynb 检索在任务配置中设置的参数 received_message 的值。

# This notebook receives the message as a parameter.

dbutils.widgets.text("received_message", "")
received_message = dbutils.widgets.get("received_message")

print(f"Received message: {received_message}")

任务设置

本部分包含每个作业任务类型的设置和示例。

条件任务

你可以通过 condition_task 向作业添加具有 if/else 条件逻辑的任务。 该任务评估可用于控制其他任务执行的条件。 条件任务不需要群集来执行，也不支持重试或通知。 有关 if/else 条件任务的详细信息，请参阅 使用 If/else 任务向作业添加分支逻辑。

以下键可用于条件任务。 有关相应的 REST API 对象定义，请参阅 condition_task。

例子

以下示例包含条件任务和笔记本任务，其中笔记本任务仅在作业修复数小于 5 时执行。

resources:
  jobs:
    my-job:
      name: my-job
      tasks:
        - task_key: condition_task
          condition_task:
            op: LESS_THAN
            left: '{{job.repair_count}}'
            right: '5'
        - task_key: notebook_task
          depends_on:
            - task_key: condition_task
              outcome: 'true'
          notebook_task:
            notebook_path: ../src/notebook.ipynb

控制面板任务

使用此任务刷新仪表板并将快照发送到订阅者。 有关捆绑包中的仪表板的详细信息，请参阅 仪表板。

以下密钥可用于 仪表板任务。 有关相应的 REST API 对象定义，请参阅 dashboard_task。

例子

以下示例将仪表板中的任务添加到作业。 运行作业时，将刷新具有指定 ID 的仪表板。

resources:
  jobs:
    my-dashboard-job:
      name: my-dashboard-job
      tasks:
        - task_key: my-dashboard-task
          dashboard_task:
            dashboard_id: 11111111-1111-1111-1111-111111111111

dbt 任务

使用此任务运行一个或多个 dbt 命令。 有关 dbt 的详细信息，请参阅 “连接到 dbt Cloud”。

以下键可用于 dbt 任务。 有关相应的 REST API 对象定义，请参阅 dbt_task。

例子

以下示例向作业添加 dbt 任务。 此 dbt 任务使用指定的 SQL 仓库来运行指定的 dbt 命令。

若要获取 SQL 仓库的 ID，请打开 SQL 仓库的设置页，然后复制“概述”选项卡上“名称”字段中仓库名称后面括号中的 ID。

resources:
  jobs:
    my-dbt-job:
      name: my-dbt-job
      tasks:
        - task_key: my-dbt-task
          dbt_task:
            commands:
              - 'dbt deps'
              - 'dbt seed'
              - 'dbt run'
            project_directory: /Users/someone@example.com/Testing
            warehouse_id: 1a111111a1111aa1
          libraries:
            - pypi:
                package: 'dbt-databricks>=1.0.0,<2.0.0'

For each 任务

你可以通过 for_each_task 向作业添加包含 for each 循环的任务。 该任务为提供的每个输入执行嵌套任务。 如需获取有关 for_each_task的详细信息，请参阅 。使用 For each 任务在循环中运行另一个任务。

以下键可用于for_each_task。 有关相应的 REST API 对象定义，请参阅 for_each_task。

例子

以下示例向作业添加一个 for_each_task，在该作业中循环访问另一个任务的值并将其处理。

resources:
  jobs:
    my_job:
      name: my_job
      tasks:
        - task_key: generate_countries_list
          notebook_task:
            notebook_path: ../src/generate_countries_list.ipnyb
        - task_key: process_countries
          depends_on:
            - task_key: generate_countries_list
          for_each_task:
            inputs: '{{tasks.generate_countries_list.values.countries}}'
            task:
              task_key: process_countries_iteration
              notebook_task:
                notebook_path: ../src/process_countries_notebook.ipnyb

JAR 任务

使用此任务运行 JAR。 可以引用本地 JAR 库或工作区、Unity Catalog 卷或外部云存储位置中的 JAR 库。 请参阅 JAR 文件（Java 或 Scala）。

若要详细了解如何在标准访问模式下在启用了 Unity Catalog 的群集上编译和部署 Scala JAR 文件，请参阅在 Unity Catalog 群集上部署 Scala JAR。

以下密钥可用于 JAR 任务。 有关相应的 REST API 对象定义，请参阅 jar_task。

例子

以下示例向作业添加 JAR 任务。 JAR 文件的路径指向一个卷位置。

resources:
  jobs:
    my-jar-job:
      name: my-jar-job
      tasks:
        - task_key: my-jar-task
          spark_jar_task:
            main_class_name: org.example.com.Main
          libraries:
            - jar: /Volumes/main/default/my-volume/my-project-0.1.0-SNAPSHOT.jar

笔记本任务

使用此任务运行笔记本。 请参阅作业的笔记本任务。

以下密钥可用于笔记本任务。 有关相应的 REST API 对象定义，请参阅 notebook_task。

例子

以下示例将笔记本任务添加到作业，并设置名为 my_job_run_id 的作业参数。 要部署的笔记本的路径是相对于声明了此任务的配置文件而言。 该任务从其在 Azure Databricks 工作区中的部署位置获取笔记本。

resources:
  jobs:
    my-notebook-job:
      name: my-notebook-job
      tasks:
        - task_key: my-notebook-task
          notebook_task:
            notebook_path: ./my-notebook.ipynb
      parameters:
        - name: my_job_run_id
          default: '{{job.run_id}}'

管道任务

你可以用这个任务来运行一个管道。 请参阅 Lakeflow 声明性管道。

以下密钥可用于 管道任务。 有关相应的 REST API 对象定义，请参阅 pipeline_task。

例子

以下示例将流水线任务添加到作业中。 此任务运行指定的管道。

resources:
  jobs:
    my-pipeline-job:
      name: my-pipeline-job
      tasks:
        - task_key: my-pipeline-task
          pipeline_task:
            pipeline_id: 11111111-1111-1111-1111-111111111111

Python 脚本任务

使用此任务运行 Python 文件。

以下密钥可用于 Python 脚本任务。 有关相应的 REST API 对象定义，请参阅 python_task。

例子

以下示例向作业添加 Python 脚本任务。 要部署的 Python 文件的路径是相对于声明了此任务的配置文件而言。 该任务从其在 Azure Databricks 工作区中的部署位置获取 Python 文件。

resources:
  jobs:
    my-python-script-job:
      name: my-python-script-job

      tasks:
        - task_key: my-python-script-task
          spark_python_task:
            python_file: ./my-script.py

Python wheel 任务

使用此任务运行 Python 滚轮。 请参阅 使用 Databricks 资产捆绑包生成 Python 滚轮文件。

以下键可用于 Python 滚轮任务。 有关相应的 REST API 对象定义，请参阅 python_wheel_task。

例子

以下示例向作业添加 Python wheel 任务。 要部署的 Python wheel 文件的路径是相对于声明了此任务的配置文件而言。 请参阅 Databricks 资产捆绑包的库依赖项。

resources:
  jobs:
    my-python-wheel-job:
      name: my-python-wheel-job
      tasks:
        - task_key: my-python-wheel-task
          python_wheel_task:
            entry_point: run
            package_name: my_package
          libraries:
            - whl: ./my_package/dist/my_package-*.whl

运行作业任务

使用此任务运行另一个作业。

以下密钥可用于 运行作业任务。 有关相应的 REST API 对象定义，请参阅 run_job_task。

例子

以下示例在第二个作业中包含了运行第一个作业的运行作业任务。

此示例使用替换来检索要运行的作业的 ID。 若要从 UI 获取作业的 ID，请在工作区中打开作业，并从作业“设置”页的“作业详细信息”选项卡中的“作业 ID”值中复制该 ID。

resources:
  jobs:
    my-first-job:
      name: my-first-job
      tasks:
        - task_key: my-first-job-task
          new_cluster:
            spark_version: '13.3.x-scala2.12'
            node_type_id: 'i3.xlarge'
            num_workers: 2
          notebook_task:
            notebook_path: ./src/test.py
    my_second_job:
      name: my-second-job
      tasks:
        - task_key: my-second-job-task
          run_job_task:
            job_id: ${resources.jobs.my-first-job.id}

SQL 任务

使用此任务运行 SQL 文件、查询或警报。

以下密钥可用于 SQL 任务。 有关相应的 REST API 对象定义，请参阅 sql_task。

例子

以下示例向作业添加 SQL 文件任务。 此 SQL 文件任务使用指定的 SQL 仓库来运行指定的 SQL 文件。

resources:
  jobs:
    my-sql-file-job:
      name: my-sql-file-job
      tasks:
        - task_key: my-sql-file-task
          sql_task:
            file:
              path: /Users/someone@example.com/hello-world.sql
              source: WORKSPACE
            warehouse_id: 1a111111a1111aa1

以下示例将 SQL 警报任务添加到作业。 此 SQL 警报任务使用指定的 SQL 仓库刷新指定的 SQL 警报。

resources:
  jobs:
    my-sql-file-job:
      name: my-sql-alert-job
      tasks:
        - task_key: my-sql-alert-task
          sql_task:
            warehouse_id: 1a111111a1111aa1
            alert:
              alert_id: 11111111-1111-1111-1111-111111111111

以下示例将 SQL 查询任务添加到作业。 此 SQL 查询任务使用指定的 SQL 仓库来运行指定的 SQL 查询。

resources:
  jobs:
    my-sql-query-job:
      name: my-sql-query-job
      tasks:
        - task_key: my-sql-query-task
          sql_task:
            warehouse_id: 1a111111a1111aa1
            query:
              query_id: 11111111-1111-1111-1111-111111111111

其他任务设置

以下任务设置允许为所有任务配置行为。 有关相应的 REST API 对象定义，请参阅 任务。