AI 数据血缘自动构建：跨系统追踪字段从哪来、到哪去

AI 数据血缘自动构建：跨系统追踪字段从哪来、到哪去

一、数据血缘为什么让分析师头疼

做过数据治理的同学一定深有体会：总监问"营收看板上的 GMV 是怎么算的？"，你得从 BI 看板一路追到数仓 ODS 层，中间经过 ETL、宽表、聚合视图，至少跨四五个系统。没有血缘工具的时候，全靠人工翻 SQL 脚本，一个字段查半天。

传统血缘方案有两种，各有致命伤：

• 人工维护：让每个 ETL 开发者在建表时手动标注来源。理想丰满，现实骨感——没人会认真维护，三个月就成了一堆过时注释。
• 静态解析：用正则或 AST 解析 SQL，提取FROM、JOIN里的表和字段引用。问题是跨系统链路（MySQL → Hive → ClickHouse → BI 工具）它根本连不上，而且复杂 ETL 里大量动态 SQL、Python UDF 之类根本解析不了。

AI 在这里的切入点很巧妙：不靠模板匹配，而是靠理解代码语义来追踪字段流转。

graph TD subgraph "数据源层" A1["MySQL 业务库<br/>orders 表"] A2["日志系统<br/>user_click_log"] end subgraph "ETL 处理层" B1["Spark ETL 任务<br/>order_agg.py"] B2["Hive SQL 脚本<br/>user_profile.sql"] B3["Python 数据处理<br/>user_feature.py"] end subgraph "数仓层" C1["Hive: dwd_order_daily"] C2["Hive: dim_user_info"] C3["ClickHouse: user_behavior"] end subgraph "应用层" D1["BI 看板<br/>GMV 指标"] D2["推荐模型<br/>训练数据"] end A1 -->|ODS 同步| B1 A2 -->|流式采集| B2 B1 --> C1 B2 --> C2 B3 --> C3 C1 --> D1 C2 --> D1 C3 --> D2 style B1 fill:#f9f,stroke:#333,stroke-width:2px style B2 fill:#f9f,stroke:#333,stroke-width:2px style B3 fill:#f9f,stroke:#333,stroke-width:2px

二、AI 血缘解析的核心思路

传统静态解析解决不了的问题，AI 用语义理解来补。具体来说，大模型做三件事：

1. 读懂代码在干嘛：不只看语法，而是理解这段代码"从哪个字段算出了哪个新字段"
2. 跨脚本关联：A 任务的输出表是 B 任务的输入表，AI 能识别这种依赖关系
3. 生成结构化血缘关系：输出标准的字段级血缘 JSON，而不是一段文字描述

from typing import List, Dict, Optional from dataclasses import dataclass, field @dataclass class FieldLineage: """单个字段的血缘关系""" field_name: str # 字段名 source_table: str # 来源表 source_field: str # 来源字段 transform_type: str # 转换类型：direct/passthrough/computed/aggregated transform_expr: Optional[str] = None # 转换表达式，如 "SUM(amount)" @dataclass class TableLineage: """一张表的完整血缘""" target_table: str fields: List[FieldLineage] = field(default_factory=list) upstream_tables: List[str] = field(default_factory=list) class AILineageParser: """基于 LLM 的数据血缘解析器""" def parse_sql_lineage(self, sql: str, table_name: str) -> TableLineage: """解析一段 SQL 的字段级血缘""" prompt = f"""分析以下 SQL，提取字段级数据血缘。 SQL 代码： ```sql {sql}

目标表名：{table_name}

对于目标表的每个字段，分析它的：

1. field_name：字段名
2. source_table：来源表名（如果是计算字段，列出所有参与计算的源表）
3. source_field：来源字段名（如果是表达式计算结果，说明计算逻辑）
4. transform_type：转换类型
   • "direct"：直接引用（如 SELECT a，则 a 的源头就是上游的 a 列）
   • "computed"：通过表达式计算（如 SUM(a+b)）
   • "renamed"：重命名（如 SELECT a AS b）
   • "constant"：常量或字面值

以 JSON 格式输出：
{{
"fields": [
{{"field_name": "total_gmv", "source_table": "orders",
"source_field": "amount", "transform_type": "computed",
"transform_expr": "SUM(amount)"}}
],
"upstream_tables": ["orders", "refunds"]
}}"""

response = self._call_llm(prompt) return self._parse_response(response) def trace_cross_system(self, lineage_list: List[TableLineage]) -> Dict: """跨系统追踪：把多段血缘拼接成完整链路""" # 构建表间依赖图：table_name -> [upstream_tables] dependency_graph = {} for lineage in lineage_list: dependency_graph[lineage.target_table] = lineage.upstream_tables # 对每个字段，从目标表反向追溯到原始数据源 full_lineage = {} for lineage in lineage_list: for field in lineage.fields: chain = self._trace_field_chain( field, lineage.target_table, dependency_graph, lineage_list ) full_lineage[f"{lineage.target_table}.{field.field_name}"] = chain return full_lineage def _trace_field_chain(self, field: FieldLineage, table: str, dep_graph: Dict, all_lineages: List[TableLineage]): """递归追踪字段的完整链路""" chain = [{ "table": table, "field": field.field_name, "transform": field.transform_expr or field.transform_type }] # 如果是直接引用，继续向上追踪 if field.transform_type == "direct" and field.source_table: for upstream in all_lineages: if upstream.target_table == field.source_table: # 找到上游表中对应的字段，继续递归 for uf in upstream.fields: if uf.field_name == field.source_field: chain = self._trace_field_chain( uf, upstream.target_table, dep_graph, all_lineages ) + chain return chain

## 三、多系统血缘拼接的难点和解决思路 最难的不是解析单段 SQL，而是**把 Hive、Spark、Python、ClickHouse 这些异构系统的血缘拼起来**。每段代码的输入输出通过"表名"这个桥梁关联，但不同系统的表命名规范可能不一样。 解决思路分三步： ```python class CrossSystemLinker: """跨系统血缘连接器""" def __init__(self): # 表名映射表，解决不同系统对同一张表的不同命名 # 例如 MySQL 里叫 orders，到 Hive ODS 层可能叫 ods_orders_di self.table_alias_map = {} def register_alias(self, canonical_name: str, alias: str): """注册表名别名映射""" self.table_alias_map[alias] = canonical_name def extract_lineage_from_etl_tasks(self, task_list: List[Dict]) -> List[TableLineage]: """ 批量解析多个 ETL 任务的血缘 task_list 每个元素包含: - script_path: 脚本文件路径 - script_type: sql/python/spark - output_table: 输出表名 """ results = [] for task in task_list: with open(task['script_path'], 'r') as f: script_content = f.read() if task['script_type'] == 'sql': lineage = self._parse_sql(script_content, task['output_table']) elif task['script_type'] == 'python': lineage = self._parse_python(script_content, task['output_table']) elif task['script_type'] == 'spark': lineage = self._parse_spark(script_content, task['output_table']) # 规范化表名 lineage = self._normalize_table_names(lineage) results.append(lineage) # 拼接所有血缘关系 full_graph = self._build_lineage_graph(results) return full_graph def _normalize_table_names(self, lineage: TableLineage) -> TableLineage: """将不同系统的表名统一到标准名称""" # MySQL: orders → ods.orders # Hive: ods_orders_di → ods.orders # 通过映射表统一，确保跨系统引用能匹配上 for alias, canonical in self.table_alias_map.items(): if lineage.target_table == alias: lineage.target_table = canonical return lineage

四、实际效果和适用边界

AI 血缘解析不是银弹，它有明确的适用场景和限制：

好用的情况：

• 标准 SQL ETL（Hive、Spark SQL、ClickHouse 查询），正确率 85%+
• Python 数据处理脚本中的 DataFrame 操作（pandas、PySpark），正确率 75%+
• 字段名和表名命名规范、注释清晰的项目

翻车的情况：

• 大量动态执行的 SQL（字符串拼接、mybatis 模板），AI 只能看模板猜结构
• 过于复杂的 UDF 和自定义函数，AI 可能理解错计算逻辑
• 没有表结构 Schema 信息，AI 不知道字段的数据类型和含义

推荐的使用姿势是半自动模式：AI 生成初版血缘后，人工 review 关键链路（尤其是面向管理层的核心指标）。这样既保证了效率，又兜底了准确性。我们在实际落地中发现，半自动模式比纯人工维护的效率提高了约5倍，review时间从按天计变成了按小时计。

五、总结

AI 做数据血缘最大的价值在于补上了静态解析搞不定的语义理解：

• AI 能读懂 SQL 和数据处理代码的意图，而不只是正则匹配 FROM 子句。对于计算字段、UDF、复杂 JOIN 这类场景，语义理解的正确率远超规则匹配。
• 跨系统拼接的核心是表名统一：MySQL、Hive、ClickHouse 对同一张表的命名可能不同，建立别名映射表是跨系统追踪的基础。
• 字段级血缘比表级血缘实用得多：知道"GMV 这个指标最终来自 orders.amount"比知道"这张表依赖哪些表"更有价值，也更能帮分析师定位问题。
• 半自动 > 全自动：AI 生成的初版血缘 + 人工 review 核心链路，是当前最务实的方案。让 AI 处理 80% 的简单字段，人工把关 20% 的关键指标。

数据治理是个长线工程，AI 让"查一个字段的来龙去脉"从几个小时变成了几分钟。这个价值，做数据的都懂。

最后提醒一点：这个方案在上生产之前建议先用灰度流量验证一周，确认资源消耗在预期范围内再全量推送。我们在实际项目中因为跳过了这步，有一次把缓存集群打挂了，教训深刻。