第20课：数据血缘

🔄 数据血缘

数据血缘追踪数据的来源、转换和流向，支持影响分析、合规审计和根因定位。

🐍 Python实战：核心实战

# 数据血缘追踪引擎
import re, json
from datetime import datetime
from collections import defaultdict

class LineageGraph:
    def __init__(self): self.nodes = {}; self.edges = []
    def add_node(self, name, node_type='table', metadata=None):
        self.nodes[name] = {'type':node_type,'metadata':metadata or {},'created':datetime.now().isoformat()}
    def add_edge(self, source, target, transform=''):
        self.edges.append({'source':source,'target':target,'transform':transform})
    def upstream(self, node, depth=5):
        visited = set(); result = []
        def traverse(n, d):
            if d > depth or n in visited: return
            visited.add(n)
            for e in self.edges:
                if e['target'] == n:
                    result.append({'node':e['source'],'depth':d,'transform':e['transform']})
                    traverse(e['source'], d+1)
        traverse(node, 1)
        return result
    def downstream(self, node, depth=5):
        visited = set(); result = []
        def traverse(n, d):
            if d > depth or n in visited: return
            visited.add(n)
            for e in self.edges:
                if e['source'] == n:
                    result.append({'node':e['target'],'depth':d,'transform':e['transform']})
                    traverse(e['target'], d+1)
        traverse(node, 1)
        return result
    def impact_analysis(self, node):
        ds = self.downstream(node)
        return {'source':node,'affected_nodes':len(set(r['node'] for r in ds)),'affected':sorted(set(r['node'] for r in ds))}
    def root_cause(self, node):
        us = self.upstream(node)
        return {'target':node,'root_sources':sorted(set(r['node'] for r in us if r['depth']==max(x['depth'] for x in us) if us else 0))}

class SQLLineageParser:
    def parse(self, sql):
        sql_lower = sql.lower().strip()
        sources = []; targets = []
        if sql_lower.startswith('create table') or sql_lower.startswith('create view'):
            parts = re.split(r'\s+as\s+', sql_lower, maxsplit=1)
            if len(parts) >= 2:
                target_match = re.search(r'(?:table|view)\s+(\w+\.?\w*)', parts[0])
                if target_match: targets.append(target_match.group(1))
                from_match = re.findall(r'from\s+(\w+\.?\w*)', parts[1])
                join_match = re.findall(r'join\s+(\w+\.?\w*)', parts[1])
                sources = from_match + join_match
        return {'sources':sources,'targets':targets}

# 构建血缘图
graph = LineageGraph()
for node in ['mysql.orders','api.products','raw_orders','raw_products',
             'stg_orders','stg_products','int_order_items',
             'fct_sales','dim_product','ads_daily_report','ads_product_rank']:
    graph.add_node(node)

graph.add_edge('mysql.orders','raw_orders','CDC同步')
graph.add_edge('api.products','raw_products','API拉取')
graph.add_edge('raw_orders','stg_orders','清洗标准化')
graph.add_edge('raw_products','stg_products','清洗标准化')
graph.add_edge('stg_orders','int_order_items','关联+衍生')
graph.add_edge('stg_products','int_order_items','Join')
graph.add_edge('int_order_items','fct_sales','聚合')
graph.add_edge('stg_products','dim_product','维度加工')
graph.add_edge('fct_sales','ads_daily_report','汇总')
graph.add_edge('dim_product','ads_product_rank','排名')
graph.add_edge('fct_sales','ads_product_rank','聚合')

# 血缘查询
print("数据血缘引擎:")
print(f"\n上游(ads_daily_report):")
for item in graph.upstream('ads_daily_report'):
    print(f"  {'  '*item['depth']}← {item['node']} ({item['transform']})")

print(f"\n下游(mysql.orders):")
for item in graph.downstream('mysql.orders'):
    print(f"  {'  '*item['depth']}→ {item['node']} ({item['transform']})")

print(f"\n影响分析(raw_orders): {graph.impact_analysis('raw_orders')}")

# SQL血缘解析
parser = SQLLineageParser()
sql = "CREATE TABLE fct_sales AS SELECT o.id, o.amount FROM stg_orders o JOIN stg_products p ON o.product_id = p.id"
result = parser.parse(sql)
print(f"\nSQL解析: {result}")
print("✅ 验证通过 - 血缘追踪引擎运行正常")

📖 原理深入：数据编排与治理体系

🏆 最佳实践：数据编排与治理检查清单

🔬 深度案例：生产环境实战经验

📚 延伸阅读与参考

🤔 常见问题与解答