如何在Databricks作业运行时获取其精确运行链接URL？

核心思路 #

问题根源在于databricks runs submit --wait会阻塞到作业完全结束后才返回结果，导致你只能在作业完成后拿到job_id和run_id。解决办法是拆分「提交作业」和「等待完成」两个独立操作：先提交作业并立即获取运行元数据（包含job_id和run_id），生成精确链接后再等待作业完成，既不影响后续预处理流程，又能在作业运行时拿到目标链接。

方案1：使用Databricks CLI拆分操作 #

修改原有脚本，分三步执行：

1. 不带--wait参数提交作业，立即获取运行元数据
2. 提取job_id和run_id生成精确运行链接
3. 调用databricks runs wait等待作业完成，再执行后续预处理

示例代码：

import json
import subprocess

# 1. 提交作业（不带--wait，立即返回运行元数据）
submit_body = {
    'run_name': 'run_name',
    'libraries': [],
    'spark_python_task': {
        'python_file': 'python file location',
        'parameters': [
            '--databricks-host', 'your-databricks-host',
            '--other-params', '...'
            # 无需提前传入{{job_id}}/{{run_id}}，将从提交结果中提取
        ]
    }
}
submit_cmd = ["databricks", "runs", "submit", "--version", "2.1", "--json", json.dumps(submit_body)]
submit_result = subprocess.run(submit_cmd, capture_output=True, text=True)

# 2. 解析结果生成精确链接
if submit_result.returncode == 0:
    run_metadata = json.loads(submit_result.stdout)
    job_id = run_metadata['job_id']
    run_id = run_metadata['run_id']
    host = 'your-databricks-host'
    run_url = f"{host}#job/{job_id}/run/{run_id}"
    print(f"作业已启动，运行链接：{run_url}")

    # 3. 等待作业完成，执行后续预处理
    wait_cmd = ["databricks", "runs", "wait", str(run_id)]
    wait_result = subprocess.run(wait_cmd, capture_output=True, text=True)
    
    if wait_result.returncode == 0:
        print("作业完成，开始后续预处理...")
        # 在这里添加你的预处理逻辑
else:
    print(f"提交作业失败：{submit_result.stderr}")

方案2：直接调用Databricks REST API #

如果更倾向于用代码而非CLI，可直接调用Databricks的REST API，提交后立即从响应中提取job_id和run_id，再调用等待接口确保作业完成。

示例代码：

import requests
import json

host = 'your-databricks-host'
token = 'your-databricks-token'
headers = {'Authorization': f'Bearer {token}', 'Content-Type': 'application/json'}

# 1. 提交作业并获取运行元数据
submit_url = f"{host}/api/2.1/jobs/runs/submit"
submit_body = {
    'run_name': 'run_name',
    'libraries': [],
    'spark_python_task': {
        'python_file': 'python file location',
        'parameters': [
            '--databricks-host', host,
            '--other-params', '...'
        ]
    }
}

response = requests.post(submit_url, headers=headers, json=submit_body)
if response.status_code == 200:
    run_metadata = response.json()
    job_id = run_metadata['job_id']
    run_id = run_metadata['run_id']
    run_url = f"{host}#job/{job_id}/run/{run_id}"
    print(f"作业已启动，运行链接：{run_url}")

    # 2. 等待作业完成
    wait_url = f"{host}/api/2.1/jobs/runs/wait"
    wait_body = {'run_id': run_id}
    wait_response = requests.post(wait_url, headers=headers, json=wait_body)
    
    if wait_response.status_code == 200:
        print("作业完成，开始后续预处理...")
        # 在这里添加你的预处理逻辑
else:
    print(f"提交作业失败：{response.text}")

注意事项 #

• 若作业基于现有Job模板提交（非一次性作业），job_id会是模板的固定ID，run_id是本次运行的唯一ID，链接格式依然有效。
• 确保你的Databricks CLI或API账号拥有「提交作业」和「查询运行状态」的权限。

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