Netflix 用一张图管住上千个模型：Model Lifecycle Graph 实践解析

# ml_lifecycle_graph.py
# 依赖: pip install networkx matplotlib pyyaml

import networkx as nx
import yaml
from typing import Dict, Any

# ---------- 1. 定义节点与边的类型标签 ----------
NODE_TYPES = {"dataset", "feature", "model", "workflow"}
EDGE_LABELS = {
    ("feature", "dataset"): "derived_from",
    ("model", "feature"):    "consumes",
    ("workflow", "model"):   "produces",
    ("workflow", "dataset"): "reads",
    ("workflow", "feature"): "computes",
}

# ---------- 2. 从 YAML 声明式配置加载图 ----------
def load_graph_from_yaml(path: str) -> nx.DiGraph:
    """把 YAML 里声明的实体和依赖关系加载为有向图"""
    with open(path) as f:
        spec = yaml.safe_load(f)

    G = nx.DiGraph()
    for node in spec.get("nodes", []):
        assert node["type"] in NODE_TYPES, f"未知节点类型: {node['type']}"
        G.add_node(node["id"], type=node["type"], meta=node.get("meta", {}))

    for edge in spec.get("edges", []):
        src, dst = edge["from"], edge["to"]
        src_type = G.nodes[src]["type"]
        dst_type = G.nodes[dst]["type"]
        label = EDGE_LABELS.get((src_type, dst_type), "depends_on")
        G.add_edge(src, dst, label=label)

    return G

# ---------- 3. 核心查询：血缘追踪 ----------
def upstream_lineage(G: nx.DiGraph, node_id: str, depth: int = 10) -> list:
    """沿入边回溯，返回某节点的全部上游依赖链"""
    lineage = []
    for u, v in nx.bfs_edges(G.reverse(), source=node_id, depth_limit=depth):
        lineage.append({
            "from": u,
            "to": v,
            "edge_label": G.edges[u, v]["label"],
            "from_type": G.nodes[u]["type"],
        })
    return lineage

def downstream_impact(G: nx.DiGraph, node_id: str, depth: int = 10) -> list:
    """沿出边遍历，返回某节点的全部下游影响链"""
    impact = []
    for u, v in nx.bfs_edges(G, source=node_id, depth_limit=depth):
        impact.append({
            "from": u,
            "to": v,
            "edge_label": G.edges[u, v]["label"],
            "to_type": G.nodes[v]["type"],
        })
    return impact

# ---------- 4. 可视化（调试用） ----------
def draw_graph(G: nx.DiGraph):
    import matplotlib.pyplot as plt
    color_map = {
        "dataset": "#4ECDC4", "feature": "#FFE66D",
        "model": "#FF6B6B", "workflow": "#95E1D3",
    }
    colors = [color_map[G.nodes[n]["type"]] for n in G.nodes]
    nx.draw(G, with_labels=True, node_color=colors, node_size=800,
            font_size=9, arrows=True)
    plt.savefig("lifecycle_graph.png", dpi=150)
    plt.close()
    print("图已保存到 lifecycle_graph.png")

# ---------- 5. 主入口 ----------
if __name__ == "__main__":
    G = load_graph_from_yaml("lifecycle_spec.yaml")

    # 示例查询：推荐排序模型的上游血缘
    print("=== 推荐排序模型的上游血缘 ===")
    for item in upstream_lineage(G, "model:ranking_v3"):
        print(f"  {item['from']}({item['from_type'])} → {item['to']}  [{item['edge_label']}]")

    # 示例查询：用户观看数据集的下游影响
    print("\n=== 用户观看数据集的下游影响 ===")
    for item in downstream_impact(G, "dataset:watch_events"):
        print(f"  {item['from']} → {item['to']}({item['to_type']})  [{item['edge_label']}]")

    draw_graph(G)

# lifecycle_spec.yaml
nodes:
  - id: dataset:watch_events
    type: dataset
    meta:
      owner: data-team
      sensitivity: pii
      freshness: daily

  - id: dataset:content_metadata
    type: dataset
    meta:
      owner: content-team
      sensitivity: internal

  - id: feature:avg_completion_rate_7d
    type: feature
    meta:
      owner: feature-team
      description: "用户过去7天平均完播率"

  - id: feature:genre_affinity_vector
    type: feature
    meta:
      owner: feature-team
      description: "用户类型偏好向量"

  - id: workflow:daily_feature_pipeline
    type: workflow
    meta:
      schedule: "0 6 * * *"
      platform: spark

  - id: workflow:ranking_train_job
    type: workflow
    meta:
      trigger: manual
      framework: pytorch

  - id: model:ranking_v3
    type: model
    meta:
      owner: ml-reco-team
      framework: pytorch
      version: "3.2"

edges:
  - from: feature:avg_completion_rate_7d
    to:   dataset:watch_events       # 特征派生自数据集

  - from: feature:genre_affinity_vector
    to:   dataset:content_metadata

  - from: workflow:daily_feature_pipeline
    to:   dataset:watch_events       # 管线读取数据集

  - from: workflow:daily_feature_pipeline
    to:   feature:avg_completion_rate_7d  # 管线计算特征

  - from: workflow:daily_feature_pipeline
    to:   feature:genre_affinity_vector

  - from: model:ranking_v3
    to:   feature:avg_completion_rate_7d  # 模型消费特征

  - from: model:ranking_v3
    to:   feature:genre_affinity_vector

  - from: workflow:ranking_train_job
    to:   model:ranking_v3           # 管线产出模型

  - from: workflow:ranking_train_job
    to:   feature:avg_completion_rate_7d  # 训练也读特征

pip install networkx matplotlib pyyaml
python ml_lifecycle_graph.py