如何通过Azure Databricks API创建Python Wheel类型作业

我需要通过Azure Databricks API在Azure中创建Python Wheel类型的Databricks作业，用于执行已开发完成的Python Wheel包，且该作业需要7×24小时运行，开发团队要求必须使用Python Wheel类型而非Notebook类型。

我已掌握通过API创建Notebook类型作业的方法，但需要将现有代码修改为创建Python Wheel类型作业，以实现DevOps流水线的自动化部署。以下是修改后的完整代码：

import requests
import os

# 使用2.1版本API创建作业
dbrks_create_job_url = f"https://{os.environ['DBRKS_INSTANCE']}.azuredatabricks.net/api/2.1/jobs/create"

DBRKS_REQ_HEADERS = {
    'Authorization': f"Bearer {os.environ['DBRKS_BEARER_TOKEN']}",
    'X-Databricks-Azure-Workspace-Resource-Id': f"/subscriptions/{os.environ['DBRKS_SUBSCRIPTION_ID']}/resourceGroups/{os.environ['DBRKS_RESOURCE_GROUP']}/providers/Microsoft.Databricks/workspaces/{os.environ['DBRKS_WORKSPACE_NAME']}",
    'X-Databricks-Azure-SP-Management-Token': os.environ['DBRKS_MANAGEMENT_TOKEN']
}

CLUSTER_ID = os.environ["DBRKS_CLUSTER_ID"]
# 配置Python Wheel的关键参数：包名、入口点、参数（按需调整）
WHEEL_PACKAGE_NAME = os.environ.get("WHEEL_PACKAGE_NAME", "your-wheel-package-name")
WHEEL_ENTRY_POINT = os.environ.get("WHEEL_ENTRY_POINT", "your_entry_function")
WHEEL_PARAMETERS = os.environ.get("WHEEL_PARAMETERS", "").split()  # 按空格分割参数

print(f"Target cluster ID: {CLUSTER_ID}")
print(f"Wheel package: {WHEEL_PACKAGE_NAME}, entry point: {WHEEL_ENTRY_POINT}")

body_json = {
    "name": "Run_Python_Wheel_Job",
    "tasks": [
        {
            "task_key": "ExecutePythonWheel",
            "description": "Execute Python Wheel package in automated pipeline",
            "depends_on": [],
            "existing_cluster_id": CLUSTER_ID,
            "python_wheel_task": {
                "package_name": WHEEL_PACKAGE_NAME,
                "entry_point": WHEEL_ENTRY_POINT,
                "parameters": WHEEL_PARAMETERS
            },
            "timeout_seconds": 86400,  # 调整为适合7×24运行的超时时间
            "max_retries": 3,
            "min_retry_interval_millis": 60000,
            "retry_on_timeout": True
        }
    ],
    "email_notifications": {},
    "max_concurrent_runs": 1
}

print("Request body in json format:")
print(body_json)

# 使用json参数传递请求体，避免字符串拼接的问题
response = requests.post(dbrks_create_job_url, headers=DBRKS_REQ_HEADERS, json=body_json)

if response.status_code == 200:
    print("Job created successfully!")
    print(f"Status code: {response.status_code}")
    print(f"Response content: {response.content}")
    job_id = response.json()['job_id']
    print(f"Job Id = {job_id}")
    print(f"##vso[task.setvariable variable=DBRKS_JOB_ID;isOutput=true;]{job_id}")
else:
    print("Job creation failed!")
    raise Exception(f"Error response: {response.content}")

关键修改说明 #

• 替换任务类型：将原代码中的notebook_task替换为python_wheel_task，这是Python Wheel作业对应的API任务类型。
• 配置Wheel参数：添加package_name（Wheel包的名称）、entry_point（Wheel包中定义的入口函数/脚本），可选的parameters（传递给入口点的命令行参数）。
• 优化请求体构造：改用Python字典构造请求体，再通过json参数传递，避免原代码中字符串拼接导致的语法错误。
• 调整运行参数：根据7×24运行的需求，修改timeout_seconds（设置为24小时）、重试策略等参数，确保作业稳定性。
• 环境变量扩展：新增WHEEL_PACKAGE_NAME、WHEEL_ENTRY_POINT等环境变量，便于在DevOps流水线中灵活配置。

内容的提问来源于stack exchange，提问作者E. Erfan