本地Python调用Databricks Notebook传参及请求格式错误排查

核心问题分析及修复方案 #

你遇到的MALFORMED_REQUEST错误基本是因为请求体JSON结构不符合Databricks API规范，或是缺少必填字段。以下是具体问题点和修正后的实现方案：

1. 常见错误诱因 #

• 请求体不是合法的JSON对象（Map），比如误传数组或字段嵌套格式错误
• 缺少run_name、tasks、集群配置等必填字段
• base_parameters未采用键值对Map格式传递参数
• 未设置正确的Content-Type: application/json请求头

2. 修复后的完整代码示例 #

import requests
import time

# 配置Databrics基础信息
DATABRICKS_HOST = "https://<你的Workspace实例地址>"
DATABRICKS_TOKEN = "<你的认证Token>"
REQUEST_HEADERS = {
    "Authorization": f"Bearer {DATABRICKS_TOKEN}",
    "Content-Type": "application/json"
}

# 1. 构建符合规范的submit_run请求体
submit_payload = {
    "run_name": "本地调用反转字符串任务",
    "tasks": [
        {
            "task_key": "reverse_string_task",
            "notebook_task": {
                "notebook_path": "/Workspace/你的Notebook完整路径",  # 替换为实际路径
                "base_parameters": {
                    "input_str": "Hello Databricks"  # 传入的字符串参数
                }
            },
            # 可选：使用现有集群（替换为你的集群ID）
            # "existing_cluster_id": "<你的集群ID>",
            # 或创建临时测试集群
            "new_cluster": {
                "spark_version": "13.3.x-scala2.12",
                "node_type_id": "Standard_DS3_v2",
                "num_workers": 1
            }
        }
    ]
}

# 2. 提交任务
submit_response = requests.post(
    f"{DATABRICKS_HOST}/api/2.1/jobs/runs/submit",
    json=submit_payload,  # 用json参数自动序列化，避免手动转JSON出错
    headers=REQUEST_HEADERS
)

if submit_response.status_code != 200:
    print(f"任务提交失败: {submit_response.text}")
    exit()

run_id = submit_response.json()["run_id"]
print(f"任务已提交，Run ID: {run_id}")

# 3. 轮询任务状态，直到执行完成
while True:
    status_response = requests.get(
        f"{DATABRICKS_HOST}/api/2.1/jobs/runs/get",
        params={"run_id": run_id},
        headers=REQUEST_HEADERS
    )
    life_cycle_state = status_response.json()["state"]["life_cycle_state"]
    
    if life_cycle_state in ["TERMINATED", "SKIPPED", "INTERNAL_ERROR"]:
        break
    print(f"当前任务状态: {life_cycle_state}，等待中...")
    time.sleep(5)

# 4. 获取执行结果
output_response = requests.get(
    f"{DATABRICKS_HOST}/api/2.1/jobs/runs/get-output",
    params={"run_id": run_id},
    headers=REQUEST_HEADERS
)

output_data = output_response.json()
if "error" in output_data:
    print(f"任务执行出错: {output_data['error']}")
else:
    # 提取Notebook返回的结果（需在Databricks Notebook中用dbutils.notebook.exit返回）
    reversed_result = output_data["notebook_output"]["result"]
    print(f"反转后的字符串: {reversed_result}")

3. Databricks Notebook端适配代码 #

你的Databricks Notebook需要接收参数并返回处理结果，示例代码如下：

# 读取传入的参数
input_str = dbutils.widgets.get("input_str")

# 执行字符串反转逻辑
reversed_str = input_str[::-1]

# 将结果返回给调用方
dbutils.notebook.exit(reversed_str)

4. 关键注意事项 #

• 确保notebook_path是Databricks Workspace中的完整路径（比如/Users/你的邮箱/反转字符串处理Notebook）
• 使用现有集群时，需确认集群处于运行状态，且Token具备该集群的访问权限
• 优先使用requests.post的json参数自动序列化字典，避免手动json.dumps导致的格式转义错误

内容的提问来源于stack exchange，提问作者J. Doe