调用Databricks Jobs API获取元数据时遇JSONDecodeError求助

问题根源 #

你遇到的JSONDecodeError是因为API返回的不是合法JSON内容——大概率是请求地址错误、权限不足或参数无效，导致返回了HTML错误页或者空响应，没法解析成JSON。

排查步骤 #

1. 修正baseURI格式：你写的'https:xxxxx'不完整，必须是带//的完整工作区地址，比如：

   • Azure：https://<工作区名称>.azuredatabricks.net
   • AWS：https://<账户ID>.cloud.databricks.com
   • GCP：https://<区域>.gcp.databricks.com

2. 先看响应状态和原始内容：别直接调用.json()，先加两行代码确认请求结果：

   print("状态码:", response.status_code)
   print("响应内容:", response.text)
   

   • 要是状态码是401：说明API Token无效或者没权限，检查Token的jobs.read权限
   • 要是404：要么endpoint写错了，要么job_id不存在
   • 要是200但内容是空：job_id可能有误

3. 确认API Token权限：确保你的Token拥有jobs.read权限，否则会返回权限错误页面。

修正后的单个作业查询代码 #

import requests
import json

# 替换成你的完整工作区地址
baseURI = 'https://<your-databricks-workspace-url>'
apiToken = 'xxxxxx'
job_id = '981462083700076'
endpoint = '/api/2.1/jobs/get'
headers = {'Authorization': f'Bearer {apiToken}'}

# 拼接完整请求URL
full_url = f"{baseURI}{endpoint}?job_id={job_id}"
response = requests.get(full_url, headers=headers)

# 先判断请求是否成功
if response.status_code == 200:
    job_info = response.json()
    print(json.dumps(job_info, indent=2))
else:
    print(f"请求失败，状态码: {response.status_code}")
    print(f"错误内容: {response.text}")

批量获取所有部门作业详情的方案 #

jobs/get只能查单个作业，要批量获取所有作业的名称、创建者、Notebook路径，得先用jobs/list拿到所有作业ID，再逐个查询详情：

import requests
import json

baseURI = 'https://<your-databricks-workspace-url>'
apiToken = 'xxxxxx'
headers = {'Authorization': f'Bearer {apiToken}'}

# 第一步：获取所有作业列表
list_url = f"{baseURI}/api/2.1/jobs/list"
list_resp = requests.get(list_url, headers=headers)

if list_resp.status_code != 200:
    print(f"拿作业列表失败: {list_resp.status_code} - {list_resp.text}")
else:
    jobs = list_resp.json().get('jobs', [])
    print(f"找到 {len(jobs)} 个作业")
    
    # 第二步：逐个拉取作业详情
    job_details = []
    get_url_template = f"{baseURI}/api/2.1/jobs/get?job_id={{}}"
    
    for job in jobs:
        job_id = job['job_id']
        get_resp = requests.get(get_url_template.format(job_id), headers=headers)
        
        if get_resp.status_code == 200:
            detail = get_resp.json()
            # 提取你需要的字段
            simplified = {
                "作业名称": detail.get('settings', {}).get('name'),
                "创建者": detail.get('creator_user_name'),
                "Notebook路径": detail.get('settings', {}).get('notebook_task', {}).get('notebook_path'),
                "作业ID": job_id
            }
            job_details.append(simplified)
        else:
            print(f"作业 {job_id} 详情拉取失败: {get_resp.status_code}")
    
    # 打印整理后的结果
    print(json.dumps(job_details, indent=2, ensure_ascii=False))

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