如何在Python中获取与Databricks作业运行相关的所有参数

1. 环境配置 #

首先安装Databricks官方Python SDK：
pip install databricks-sdk

提前配置好认证信息，可直接写入系统环境变量：

• DATABRICKS_HOST：你的Databricks工作区访问地址，格式为https://<工作区ID>.cloud.databricks.com
• DATABRICKS_TOKEN：你在Databricks中生成的个人访问令牌（PAT）

2. 核心数据获取代码 #

以下代码可直接拉取过去7天所有作业运行记录，提取你需要的全量参数：

from databricks.sdk import WorkspaceClient
import pandas as pd
from datetime import datetime, timedelta
import pytz

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

# 计算7天前的毫秒级时间戳（Databricks API使用毫秒级时间戳）
start_time_from = int((datetime.now(pytz.UTC) - timedelta(days=7)).timestamp() * 1000)

# 拉取符合时间范围的所有作业运行记录
all_runs = w.jobs.list_runs(
    start_time_from=start_time_from,
    expand_tasks=False # 若需要获取子任务详情可修改为True
)

# 提取需要的字段存入结构化列表
run_records = []
for run in all_runs:
    # 时间格式转换，可根据需要调整时区
    start_time = datetime.fromtimestamp(run.start_time / 1000, tz=pytz.timezone("Asia/Shanghai"))
    end_time = datetime.fromtimestamp(run.end_time / 1000, tz=pytz.timezone("Asia/Shanghai")) if run.end_time else None
    # 计算运行时长，单位为分钟
    duration_min = round((end_time - start_time).total_seconds() / 60, 2) if end_time else None
    
    run_records.append({
        "运行日期": start_time.strftime("%Y-%m-%d"),
        "开始时间": start_time.strftime("%Y-%m-%d %H:%M:%S"),
        "结束时间": end_time.strftime("%Y-%m-%d %H:%M:%S") if end_time else "运行中",
        "运行时长(分钟)": duration_min,
        "作业ID": run.job_id,
        "作业名称": run.run_name,
        "运行ID": run.run_id,
        "运行状态": run.state.value,
        "错误信息": run.state.error_message if run.state.error_message else "无",
        "触发方式": run.trigger.value if run.trigger else "手动触发",
        "触发人": run.creator_user_name
    })

# 转成DataFrame格式方便后续统计与报表生成
df = pd.DataFrame(run_records)

3. 报表生成操作 #

拿到结构化的DataFrame后可直接做各类统计和导出：

• 按日期分组统计每日作业运行概览：

  daily_stats = df.groupby(["运行日期", "运行状态"]).size().unstack(fill_value=0)
  

• 导出完整运行详情与统计结果到Excel报表：

  with pd.ExcelWriter("databricks周作业运行报表.xlsx") as writer:
      df.to_excel(writer, sheet_name="所有运行详情", index=False)
      daily_stats.to_excel(writer, sheet_name="每日运行统计")
  

提示：如果作业运行量级较大，可通过list_runs的page_token参数分页拉取数据，避免单次请求超时。

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