能否连接Azure中的Databricks数据库？求自动化数据获取指引

核心结论 #

可以通过Databricks REST API实现Notebooks和Workflows数据的自动化获取，这是官方推荐的稳定方案；直接连接Databricks底层数据库不被支持（内部存储结构无公开规范，易变动），不建议采用。

一、准备API访问凭证 #

• 获取你的Databricks工作区URL（格式：https://<workspace-id>.cloud.databricks.com）
• 生成个人访问令牌：
  1. 登录Databricks工作区，点击右上角用户头像 → User Settings
  2. 切换到Access Tokens标签页 → Generate New Token
  3. 设置令牌有效期，复制生成的令牌（仅能查看一次，务必妥善保存）
• 认证方式：所有API请求需携带Authorization: Bearer <你的令牌>请求头

二、获取Workflows（作业）数据 #

Workflows对应Databricks Jobs API，可获取关联集群、创建者、失败次数等核心信息：

1. 获取所有作业基础信息

发送GET请求到/api/2.1/jobs/list，返回所有作业的元数据：

curl -X GET -H "Authorization: Bearer <your-token>" https://<workspace-url>/api/2.1/jobs/list

响应字段说明：

• creator_user_name：作业创建者
• job_clusters/existing_cluster_id：关联的集群信息（前者是作业专属集群，后者是现有集群）
• job_id：作业唯一标识（用于后续查询运行历史）

2. 统计作业失败次数

发送GET请求到/api/2.1/jobs/runs/list，可指定时间范围和作业ID过滤运行记录：

curl -X GET -H "Authorization: Bearer <your-token>" \
"https://<workspace-url>/api/2.1/jobs/runs/list?job_id=<job-id>&start_time_from=<起始时间戳（毫秒）>&start_time_to=<结束时间戳（毫秒）>"

遍历响应中的runs数组，统计state.result_state为FAILED的条目数量，即为该作业在指定时间内的失败次数。

三、获取Notebooks数据 #

Notebooks信息通过Workspace API获取，关联集群需结合作业运行上下文：

1. 获取Notebook基础信息

发送GET请求到/api/2.0/workspace/get-status，指定Notebook路径：

curl -X GET -H "Authorization: Bearer <your-token>" \
"https://<workspace-url>/api/2.0/workspace/get-status?path=/path/to/target-notebook"

响应字段说明：

• created_by：Notebook创建者
• created_at：创建时间
• object_type：确认是NOTEBOOK类型

2. 获取Notebook关联集群

• 如果Notebook通过Workflows（作业）运行：从Jobs API返回的tasks.notebook_task字段中，提取cluster_key（专属集群）或existing_cluster_id（现有集群）
• 如果是交互式运行的Notebook：可通过/api/2.0/clusters/list和/api/2.0/contexts/list关联会话与集群，但通常监控重点是作业化运行的Notebook。

四、自动化数据整合示例（Python） #

用Python封装API调用，批量获取并整理数据，方便生成报告：

import requests
import datetime

# 配置参数
DATABRICKS_URL = "https://<your-workspace-url>"
TOKEN = "<your-access-token>"

def get_all_jobs():
    """获取所有作业列表"""
    url = f"{DATABRICKS_URL}/api/2.1/jobs/list"
    headers = {"Authorization": f"Bearer {TOKEN}"}
    resp = requests.get(url, headers=headers)
    resp.raise_for_status()
    return resp.json()["jobs"]

def count_job_failures(job_id, days=7):
    """统计指定作业最近N天的失败次数"""
    start_ts = (datetime.datetime.now() - datetime.timedelta(days=days)).timestamp() * 1000
    end_ts = datetime.datetime.now().timestamp() * 1000
    url = f"{DATABRICKS_URL}/api/2.1/jobs/runs/list"
    params = {"job_id": job_id, "start_time_from": start_ts, "start_time_to": end_ts}
    headers = {"Authorization": f"Bearer {TOKEN}"}
    resp = requests.get(url, headers=headers, params=params)
    resp.raise_for_status()
    runs = resp.json().get("runs", [])
    return sum(1 for run in runs if run["state"]["result_state"] == "FAILED")

def get_notebook_info(notebook_path):
    """获取Notebook基础信息"""
    url = f"{DATABRICKS_URL}/api/2.0/workspace/get-status"
    params = {"path": notebook_path}
    headers = {"Authorization": f"Bearer {TOKEN}"}
    resp = requests.get(url, headers=headers, params=params)
    resp.raise_for_status()
    return resp.json()

if __name__ == "__main__":
    # 批量获取作业数据
    jobs = get_all_jobs()
    for job in jobs:
        job_id = job["job_id"]
        creator = job["creator_user_name"]
        cluster = job.get("job_clusters") or job.get("existing_cluster_id", "无关联集群")
        failure_count = count_job_failures(job_id)
        print(f"作业ID: {job_id} | 创建者: {creator} | 关联集群: {cluster} | 最近7天失败次数: {failure_count}")
    
    # 获取单个Notebook信息
    notebook_info = get_notebook_info("/path/to/your/notebook")
    print(f"\nNotebook路径: {notebook_info['path']} | 创建者: {notebook_info['created_by']} | 创建时间: {datetime.datetime.fromtimestamp(notebook_info['created_at']/1000)}")

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