Python 中的配置

2025-06-18

重要

对 Databricks 资产捆绑包的 Python 支持以公共预览版提供。

用于 Databricks 资产捆绑包的 Python 可扩展 Databricks 资产捆绑包，因此你可以：

将作业和管道定义为 Python 代码。这些定义可以与 YAML 中定义的定义共存。
使用元数据动态创建任务或流程。请参阅使用元数据创建资源。
在捆绑部署期间修改 YAML 或 Python 中定义的作业。请参阅修改 YAML 或 Python 中定义的作业。

有关 Databricks 资产捆绑包 databricks 捆绑包的 Python 支持的参考文档，请参阅此文档 https://databricks.github.io/cli/experimental/python/。

要求

若要对 Databricks 资产捆绑包使用 Python 支持，必须首先：

安装 Databricks CLI 版本 0.248.0 或更高版本。请参阅安装或更新 Databricks CLI。
如果尚未这样做，请向 Databricks 工作区进行身份验证：
```
databricks configure
```
安装 uv。请参阅安装 uv。用于 Databricks 资产捆绑的 Python 使用 uv 创建虚拟环境并安装所需的依赖项。或者，可以使用其他工具（如 venv）配置 Python 环境。

从模板创建项目

若要为 Databricks 资产捆绑项目创建新的 Python 支持，请使用 experimental-jobs-as-code 模板初始化捆绑包：

databricks bundle init experimental-jobs-as-code

出现提示时，为项目命名（例如 my_jobs_as_code_project），并接受加入笔记本和 Python 包的选项。

现在，在新的项目文件夹中创建新的虚拟环境：

cd my_jobs_as_code_project
uv sync

默认情况下，模板包含文件中定义为 Python resources/my_jobs_as_code_project_job.py 的作业示例：

from databricks.bundles.jobs import Job


my_jobs_as_code_project_job = Job.from_dict(
    {
        "name": "my_jobs_as_code_project_job",
        "tasks": [
            {
                "task_key": "notebook_task",
                "notebook_task": {
                    "notebook_path": "src/notebook.ipynb",
                },
            },
        ],
    },
)

该 Job.from_dict 函数接受使用与 YAML 相同的格式的 Python 字典。还可以使用数据类语法构建任务：

from databricks.bundles.jobs import Job, Task, NotebookTask


my_jobs_as_code_project_job = Job(
    name="my_jobs_as_code_project_job",
    tasks=[
        Task(
            task_key="notebook_task",
            notebook_task=NotebookTask(
                notebook_path="src/notebook.ipynb",
            ),
        ),
    ],
)

Python 文件是通过在experimental部分中databricks.yml指定的入口点加载的。

experimental:
  python:
    # Activate the virtual environment before loading resources defined in
    # Python. If disabled, it defaults to using the Python interpreter
    # available in the current shell.
    venv_path: .venv
    # Functions called to load resources defined in Python.
    # See resources/__init__.py
    resources:
      - 'resources:load_resources'

默认情况下， resources/__init__.py 包含加载资源包中的所有 Python 文件的函数。

from databricks.bundles.core import (
    Bundle,
    Resources,
    load_resources_from_current_package_module,
)


def load_resources(bundle: Bundle) -> Resources:
    """
    'load_resources' function is referenced in databricks.yml and is responsible for loading
    bundle resources defined in Python code. This function is called by Databricks CLI during
    bundle deployment. After deployment, this function is not used.
    """

    # the default implementation loads all Python files in 'resources' directory
    return load_resources_from_current_package_module()

部署并运行作业或管道

若要将捆绑包部署到开发目标，请使用捆绑包项目根目录中的捆绑部署命令：

databricks bundle deploy --target dev

此命令部署为捆绑项目定义的所有内容。例如，使用默认模板创建的项目会将名为[dev yourname] my_jobs_as_code_project_job的作业部署到工作区。可以通过导航到 Databricks 工作区中的 作业和管道 找到该作业。

部署捆绑包后，可以使用捆绑包摘要命令查看已部署的所有内容：

databricks bundle summary --target dev

最后，若要运行作业或管道，请使用捆绑包运行命令：

databricks bundle run my_jobs_as_code_project_job

更新现有捆绑包

若要更新现有捆绑包，请根据模板创建项目中所述，为项目模板结构建模。可以使用 YAML 更新现有捆绑包，通过在experimental.python中添加databricks.yml部分，以包含定义为 Python 代码的作业或管道。

experimental:
  python:
    # Activate the virtual environment before loading resources defined in
    # Python. If disabled, it defaults to using the Python interpreter
    # available in the current shell.
    venv_path: .venv
    # Functions called to load resources defined in Python.
    # See resources/__init__.py
    resources:
      - 'resources:load_resources'

指定的虚拟环境必须包含已安装的 databricks-bundles PyPi 包。

pip install databricks-bundles==0.248.0

资源文件夹必须包含 __init__.py 文件：

from databricks.bundles.core import (
    Bundle,
    Resources,
    load_resources_from_current_package_module,
)


def load_resources(bundle: Bundle) -> Resources:
    """
    'load_resources' function is referenced in databricks.yml and
    is responsible for loading bundle resources defined in Python code.
    This function is called by Databricks CLI during bundle deployment.
    After deployment, this function is not used.
    """

    # default implementation loads all Python files in 'resources' folder
    return load_resources_from_current_package_module()

将现有作业转换为 Python

若要将现有作业转换为 Python，可以使用 视图作为代码 功能。请参阅以代码形式查看作业。

在 Databricks 工作流中打开作业页。
单击“立即运行”按钮左侧的垂直三点菜单，然后单击“以代码形式查看”：

选择 Python，然后选择 Databricks 资产捆绑包

以代码的形式查看，Python

单击 “复制 ”并将生成的 Python 另存为捆绑项目的资源文件夹中的 Python 文件。

小窍门

你还可以查看和复制现有作业和管道的 YAML 配置，然后直接粘贴到捆绑包配置 YAML 文件中。

使用元数据创建资源

函数的默认实现 load_resources 将加载包中的 resources Python 文件。可以使用 Python 以编程方式创建资源。例如，可以加载配置文件并在循环中创建作业：

from databricks.bundles.core import (
    Bundle,
    Resources,
    load_resources_from_current_package_module,
)


from databricks.bundles.jobs import Job


def create_job(country: str):
    my_notebook = {
        "task_key": "my_notebook",
        "notebook_task": {
            "notebook_path": "files/my_notebook.py",
        },
    }


    return Job.from_dict(
        {
            "name": f"my_job_{country}",
            "tasks": [my_notebook],
        }
    )


def load_resources(bundle: Bundle) -> Resources:
    resources = load_resources_from_current_package_module()


    for country in ["US", "NL"]:
        resources.add_resource(f"my_job_{country}", create_job(country))


    return resources

访问捆绑包变量

该 bundle 参数可用于访问捆绑包变量和部署目标：

from databricks.bundles.core import Bundle, Resources, Variable, variables

@variables
class Variables:
    warehouse_id: Variable[str]


def load_resources(bundle: Bundle) -> Resources:
    warehouse_id = bundle.resolve_variable(Variables.warehouse_id)

    ...

有关详细信息，请参阅 Resources 和 Bundle 类参考。

修改 YAML 或 Python 中定义的作业

可以在 databricks.yml 中引用作业赋值函数，这类似于加载资源的函数。此功能可以不依赖于 Python 中定义的加载作业来使用，并且可以修改 YAML 和 Python 中定义的资源。

首先，在捆绑根目录中创建mutators.py，并添加以下内容：

from dataclasses import replace


from databricks.bundles.core import Bundle, job_mutator
from databricks.bundles.jobs import Job, JobEmailNotifications


@job_mutator
def add_email_notifications(bundle: Bundle, job: Job) -> Job:
    if job.email_notifications:
        return job


    email_notifications = JobEmailNotifications.from_dict(
        {
            "on_failure": ["${workspace.current_user.userName}"],
        }
    )


    return replace(job, email_notifications=email_notifications)

现在，使用以下配置在捆绑部署期间执行 add_email_notifications 函数。此操作会更新捆绑包中定义的每个作业，还会发送电子邮件通知（如果没有）。必须在 databricks.yml 中指定 Mutator 函数，并按指定的顺序执行。作业可变器针对捆绑包中定义的每个作业执行，并且可以返回更新的副本或未修改的输入。可变器还可用于其他字段，例如配置默认作业群集或 SQL 仓库。

experimental:
  python:
    mutators:
      - 'mutators:add_email_notifications'

如果在可变器执行期间函数引发异常，捆绑部署将中止。

通过