如何通过Python脚本运行Azure Databricks Notebook？

依赖包安装 #

首先安装Databricks官方Python SDK：

pip install databricks-sdk

认证方式（二选一） #

方式1：Service Principal认证（使用client_id + secret）

from databricks.sdk import WorkspaceClient
from databricks.sdk.core import Config

# 替换为你的实际参数
config = Config(
    host="https://<你的Databricks工作区URL>",  # 格式示例：https://eastus.azuredatabricks.net
    client_id="<你的Service Principal Client ID>",
    client_secret="<你的Service Principal Secret>"
)

# 初始化Workspace客户端
w = WorkspaceClient(config=config)

方式2：个人Access Token认证

from databricks.sdk import WorkspaceClient

# 替换为你的实际参数
w = WorkspaceClient(
    host="https://<你的Databricks工作区URL>",
    token="<你的Databricks Access Token>"
)

提交Notebook运行（类ADF create_run） #

通过Jobs API提交Notebook运行任务，支持临时集群或现有集群：

import time

# 提交Notebook运行请求
run_response = w.jobs.submit(
    run_name="Python触发的Notebook任务",
    tasks=[
        {
            "task_key": "execute_notebook",
            "notebook_task": {
                "notebook_path": "/<Notebook的完整工作区路径>",  # 示例：/Users/your_email@domain.com/test_notebook
                "base_parameters": {  # 可选：传递给Notebook的参数
                    "input_param": "test_value",
                    "date_param": "2024-05-20"
                }
            },
            # 选项1：使用临时集群
            "new_cluster": {
                "spark_version": "13.3.x-scala2.12",
                "node_type_id": "Standard_DS3_v2",
                "num_workers": 1,
                "spark_conf": {
                    "spark.databricks.delta.preview.enabled": "true"
                }
            },
            # 选项2：使用现有集群（注释掉new_cluster，启用下面一行）
            # "existing_cluster_id": "<你的现有集群ID>"
        }
    ]
)

run_id = run_response.run_id
print(f"Notebook运行已提交，Run ID: {run_id}")

轮询运行状态（类ADF pipeline_runs.get） #

定期查询运行状态，直到任务完成：

# 获取初始状态
run_status = w.jobs.get_run(run_id=run_id).state.life_cycle_state

# 轮询状态直到任务结束
while run_status in ["PENDING", "RUNNING"]:
    print(f"当前运行状态: {run_status}，5秒后刷新...")
    time.sleep(5)
    run_status = w.jobs.get_run(run_id=run_id).state.life_cycle_state

# 获取最终运行结果
final_run_details = w.jobs.get_run(run_id=run_id)
print(f"\n任务运行结束")
print(f"最终生命周期状态: {final_run_details.state.life_cycle_state}")
print(f"任务结果状态: {final_run_details.state.result_state}")

# 可选：获取Notebook运行输出
# run_output = w.jobs.get_run_output(run_id=run_id)
# print(f"运行输出: {run_output}")

关键说明 #

• 工作区URL可在Databricks工作区主页的地址栏获取，格式为https://<区域>.azuredatabricks.net
• 若使用现有集群，需确保集群处于运行状态
• 状态枚举值说明：
  • PENDING：任务排队中
  • RUNNING：任务运行中
  • SUCCEEDED：任务成功完成
  • FAILED：任务执行失败
  • TERMINATED：任务被主动终止

内容的提问来源于stack exchange，提问作者Sam