调用Databricks REST API查询作业运行记录：分页Token问题求助

你遇到的分页无效、结果重复的核心原因是API参数名错误：Databricks jobs/runs/list API要求分页时传入page_token参数，而你代码中错误地使用了next_page_token作为参数名，导致每次请求都默认返回第一页数据。

具体修复步骤及优化点 #

1. 修正分页参数名
   将响应返回的next_page_token赋值给正确的请求参数page_token，而非next_page_token。

2. 添加请求错误处理
   新增response.raise_for_status()，及时捕获API请求失败的异常（如权限错误、网络问题等）。

3. 优化数据处理逻辑

   • 避免循环内重复创建并打印完整DataFrame，改为打印当前累积数据量跟踪进度；
   • 处理execution_duration的空值情况，避免int(None)引发的报错。

修改后的完整代码 #

import requests
import pandas as pd
import math
import datetime
import json

def fetch_and_process_job_runs(base_uri, api_token, params):
    endpoint = '/api/2.1/jobs/runs/list'
    headers = {'Authorization': f'Bearer {api_token}'}
    
    all_data = []  # 存储所有分页的数据
    
    while True:
        response = requests.get(base_uri + endpoint, headers=headers, params=params)
        # 检查请求是否成功，失败则抛出异常
        response.raise_for_status()
        response_json = response.json()
        
        data = []
        for run in response_json["runs"]:
            start_time_ms = run["start_time"]
            start_time_seconds = start_time_ms / 1000
            start_time_readable = datetime.datetime.fromtimestamp(start_time_seconds).strftime('%Y-%m-%d %H:%M:%S')
            
            # 处理execution_duration为空的情况，默认设为0
            exec_duration = run.get('execution_duration', 0)
            exec_duration_mins = math.ceil(int(exec_duration) / (1000 * 60))
            
            data.append({
                "job_id": run["job_id"],
                "creator_user_name": run["creator_user_name"],
                "run_name": run["run_name"],
                "run_page_url": run["run_page_url"],
                "run_id": run["run_id"],
                "execution_duration_in_mins": exec_duration_mins,
                "result_state": run["state"].get("result_state"),
                "start_time": start_time_readable
            })
        
        all_data.extend(data)
        # 打印当前进度，替代完整DataFrame输出
        print(f"已获取 {len(all_data)} 条作业运行记录")
        
        if response_json.get("has_more"):
            next_page_token = response_json.get("next_page_token")
            # 使用正确的分页参数名page_token
            params['page_token'] = next_page_token
        else:
            break
    
    df = pd.DataFrame(all_data)
    return df

# 替换为你的实际配置
now = datetime.datetime.utcnow()
yesterday = now - datetime.timedelta(days=1)
start_time_from = int(yesterday.replace(hour=0, minute=0, second=0, microsecond=0).timestamp()) * 1000
start_time_to = int(yesterday.replace(hour=23, minute=59, second=59, microsecond=999999).timestamp()) * 1000
        
params = {
     "start_time_from": start_time_from,
     "start_time_to": start_time_to,
     "expand_tasks": True
}
baseURI = 'https://adb-xxxxxxxxxxxxxx.azuredatabricks.net'
apiToken = 'xxxxxxxxxxxxxxxxxxxxxxxxxx'

try:
    result_df = fetch_and_process_job_runs(baseURI, apiToken, params)
    print("\n最终获取的作业运行记录：")
    print(result_df)
    # 可选：保存结果到CSV文件
    # result_df.to_csv('yesterday_job_runs.csv', index=False)
except requests.exceptions.RequestException as e:
    print(f"API请求失败：{e}")

额外注意事项 #

• 请确保启用start_time_from和start_time_to参数，否则会返回所有历史作业数据，而非仅昨日数据；
• 若作业数量极大，可添加limit参数控制每页返回的记录数（API默认值为25）；
• 代码使用UTC时间计算昨日区间，若你的Databricks工作区使用其他时区，需调整时间计算逻辑。

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