如何从dbt Explorer获取Lineage？求API或Python实现方案

一、通过dbt Cloud API间接获取血缘数据 #

dbt Cloud目前没有直接返回Lineage JSON的公开API，但可以通过组合以下API端点获取构建血缘所需的核心数据：

• 调用/api/v2/accounts/{account_id}/projects/{project_id}/runs/?status=success获取最新成功运行的run_id
• 通过/api/v2/accounts/{account_id}/projects/{project_id}/runs/{run_id}/artifacts/manifest.json拉取项目的manifest文件，其中包含所有模型的依赖关系（depends_on字段）和完全限定名称（fqn）

示例Python代码 #

import requests
import json

# 配置参数
account_id = 8888
project_id = 8888
target_model_fqn = "model.xxxx.yyyyy"
api_token = "your_dbt_cloud_api_token"

# 获取最新成功运行的run_id
def get_latest_success_run_id():
    url = f"https://cloud.getdbt.com/api/v2/accounts/{account_id}/projects/{project_id}/runs/?status=success"
    headers = {"Authorization": f"Token {api_token}"}
    resp = requests.get(url, headers=headers)
    runs = resp.json()["data"]
    return runs[0]["id"] if runs else None

# 拉取manifest并提取目标模型的血缘
run_id = get_latest_success_run_id()
if run_id:
    manifest_url = f"https://cloud.getdbt.com/api/v2/accounts/{account_id}/projects/{project_id}/runs/{run_id}/artifacts/manifest.json"
    manifest_resp = requests.get(manifest_url, headers={"Authorization": f"Token {api_token}"})
    manifest = manifest_resp.json()

    target_node = manifest["nodes"].get(target_model_fqn)
    if target_node:
        # 提取上游依赖
        upstream_nodes = target_node["depends_on"]["nodes"]
        # 提取下游依赖（遍历所有节点查找依赖当前模型的项）
        downstream_nodes = [
            node_id for node_id, node in manifest["nodes"].items()
            if target_model_fqn in node["depends_on"].get("nodes", [])
        ]

        lineage_data = {
            "target_model": target_model_fqn,
            "upstream": upstream_nodes,
            "downstream": downstream_nodes
        }
        print(json.dumps(lineage_data, indent=2))

二、生成同款Lineage可视化 #

拿到血缘数据后，可通过以下Python工具生成类似dbt Explorer的拓扑图：

• NetworkX：用于构建图结构，管理节点与边的关系
• PyVis：生成交互式HTML可视化，支持拖拽、缩放，风格接近dbt的Lineage界面
• Matplotlib：生成静态图片快照，适合嵌入报表

示例：用PyVis生成交互式Lineage #

from pyvis.network import Network

# 基于上述lineage_data构建图
net = Network(directed=True, height="600px", width="100%", bgcolor="#222222", font_color="white")

# 添加目标模型节点
net.add_node(target_model_fqn, label=target_model_fqn.split(".")[-1], color="#4169E1")

# 添加上游节点并建立连接
for node in upstream_nodes:
    node_label = node.split(".")[-1]
    net.add_node(node, label=node_label, color="#32CD32")
    net.add_edge(node, target_model_fqn)

# 添加下游节点并建立连接
for node in downstream_nodes:
    node_label = node.split(".")[-1]
    net.add_node(node, label=node_label, color="#FF6347")
    net.add_edge(target_model_fqn, node)

# 生成HTML文件，可直接打开或嵌入其他工具
net.write_html("dbt_lineage.html")

三、存储至PBI或其他介质 #

• PBI报表：将生成的HTML上传至云存储后，通过PBI的「Web内容」控件嵌入；或把血缘数据导出为CSV/JSON，导入PBI后使用自定义视觉对象（如Network Graph）构建可视化
• 其他存储：将lineage_data序列化为JSON文件保存到本地/数据库，或用pandas转为DataFrame后存储为Parquet/CSV格式

注意事项 #

• 确保API令牌拥有读取项目运行记录和模型信息的权限
• manifest.json仅包含成功运行后的模型状态，需保证存在最近的成功运行记录
• 若需全局血缘关系，可遍历manifest中所有节点构建完整图谱

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