如何通过API查找Azure Databricks运行中集群关联的笔记本？

要找出运行中集群关联的笔记本，仅通过get_cluster方法无法获取相关信息——集群的基础元数据不包含关联的笔记本路径。你需要结合交互式会话API和作业运行API来获取对应数据，以下是基于Python Databricks API的实现步骤：

步骤1：筛选所有运行中的集群 #

首先获取所有集群列表，过滤出状态为RUNNING的集群：

# 获取集群列表并筛选运行中的集群
all_clusters = dbx_env.cluster.list()["clusters"]
running_clusters = [cluster for cluster in all_clusters if cluster["state"] == "RUNNING"]

步骤2：获取集群关联的交互式笔记本 #

交互式笔记本通过会话（Context）与集群绑定，调用context.list方法可以获取指定集群的所有会话，从中提取笔记本路径：

for cluster in running_clusters:
    cluster_id = cluster["cluster_id"]
    cluster_name = cluster["cluster_name"]
    print(f"\n=== 集群 {cluster_name} ({cluster_id}) 关联的笔记本 ===")
    
    # 获取交互式会话关联的笔记本
    try:
        contexts = dbx_env.context.list(cluster_id)["contexts"]
        for ctx in contexts:
            if "notebook_path" in ctx:
                print(f"- 交互式笔记本：{ctx['notebook_path']}")
    except Exception as e:
        print(f"⚠️ 获取交互式会话失败：{str(e)}")

步骤3：获取集群关联的作业笔记本 #

如果有作业固定使用该集群运行，可通过作业运行API筛选出关联的作业任务，提取其中的笔记本路径：

# 获取运行在当前集群上的作业关联笔记本
    try:
        # 仅获取活跃状态的作业实例
        active_job_runs = dbx_env.jobs.runs.list(active_only=True)["runs"]
        for run in active_job_runs:
            # 判断作业是否使用当前集群
            if run.get("cluster_spec", {}).get("existing_cluster_id") == cluster_id:
                # 提取作业任务中的笔记本路径
                task = run["tasks"][0]
                if "notebook_task" in task:
                    notebook_path = task["notebook_task"]["notebook_path"]
                    print(f"- 作业关联笔记本：{notebook_path}（作业ID：{run['job_id']}）")
    except Exception as e:
        print(f"⚠️ 获取作业运行实例失败：{str(e)}")

注意事项 #

• 运行脚本的账号需要具备cluster:read、context:read、jobs:read权限，否则会返回权限错误。
• 若集群上没有活跃的交互式会话或作业运行，会返回空结果。

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