如何使用Azure Databricks API提交任务及现有集群调用方法

1. 获取job id后无输出的原因与解决方案 #

你调用的POST /api/2.0/jobs/create接口仅创建作业的静态定义，不会主动触发作业运行，因此没有运行产出的日志和结果。你需要按以下步骤执行才能拿到输出：

步骤1：触发作业运行，获取run_id #

import requests
import time

DOMAIN = "你的Databricks工作区域名"
TOKEN = "你的访问令牌"
JOB_ID = "你刚才创建得到的job_id"

# 触发作业运行
run_response = requests.post(
    f"https://{DOMAIN}/api/2.0/jobs/run-now",
    headers={"Authorization": f"Bearer {TOKEN}"},
    json={"job_id": JOB_ID}
)
run_id = run_response.json()["run_id"]

步骤2：轮询运行状态，完成后获取输出 #

# 轮询运行状态
while True:
    status_response = requests.get(
        f"https://{DOMAIN}/api/2.0/jobs/runs/get",
        headers={"Authorization": f"Bearer {TOKEN}"},
        params={"run_id": run_id}
    )
    status = status_response.json()["state"]["life_cycle_state"]
    if status in ["TERMINATED", "SKIPPED", "INTERNAL_ERROR"]:
        break
    print("任务运行中，等待10秒后重试...")
    time.sleep(10)

# 获取运行输出
output_response = requests.get(
    f"https://{DOMAIN}/api/2.0/jobs/runs/get-output",
    headers={"Authorization": f"Bearer {TOKEN}"},
    params={"run_id": run_id}
)
print(output_response.json())

2. 基于已有集群提交任务的代码实现 #

仅需要把原作业创建接口的new_cluster字段替换为existing_cluster_id字段，填入你的目标集群ID即可，集群ID可在Databricks控制台集群详情页的URL中获取，格式通常为数字-数字-随机字符串，示例代码如下：

import requests

DOMAIN = "你的Databricks工作区域名"
TOKEN = "你的访问令牌"
EXISTING_CLUSTER_ID = "你的已有集群ID"

response = requests.post(
  f"https://{DOMAIN}/api/2.0/jobs/create",
  headers={'Authorization': f'Bearer {TOKEN}'},
  json={
       "name": "SparkPi spark-submit job",
       "existing_cluster_id": EXISTING_CLUSTER_ID,
       "spark_submit_task": {
       "parameters": [
           "--class",
            "org.apache.spark.examples.SparkPi",
            "dbfs:/FileStore/sparkpi_assembly_0_1.jar",
            "10"
            ]
        }
    }
)
if response.status_code == 200:
  print(response.json())
else:
  print(f"Error creating job: {response.json()['error_code']}: {response.json()['message']}")

注意事项 #

• 你的访问令牌需要对应分配作业创建、集群操作、运行日志读取的权限，否则会触发权限错误
• 原代码中使用的7.3.x-scala2.12Spark版本已停止官方支持，建议替换为当前LTS版本，例如13.3.x-scala2.12
• 提交任务前请确认对应jar包已经上传到你填写的DBFS路径，否则任务运行会报文件不存在错误

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