3. 血缘采集技术：三种主流方式

血缘分析的第一步，也是最关键的一步，就是数据采集。说白了，你得先把数据从哪里来、经过什么处理、最终流向哪里，这些信息给抓取出来。

我个人习惯把血缘采集分成三个层次：表级血缘、字段级血缘和ETL任务血缘。每个层次的采集技术都不一样，踩过的坑也各不相同。

3.1 解析SQL获取表级血缘

表级血缘是最基础的。你只需要知道某张表依赖了哪些上游表，或者被哪些下游表引用。实现方式也最直接——解析SQL语句。

具体怎么做？我一般分三步走：

1. SQL抓取：从数据仓库的查询日志、调度系统的执行记录、甚至代码仓库里把SQL捞出来。
2. 语法解析：用SQL解析器（比如Antlr、sqlparse）把SQL拆成抽象语法树（AST）。
3. 表名提取：从AST中找出FROM、JOIN、INSERT INTO、CREATE TABLE AS等关键字后面的表名。

嗯，这里要注意。不是所有SQL都规规矩矩写的。我在项目中遇到过一种情况：有人把表名写在子查询的别名里，或者用CTE（公用表表达式）层层嵌套。解析器一不小心就会漏掉。

下面是一个简单的SQL解析示例（Python + sqlparse）：

import sqlparse
from sqlparse.sql import Identifier, IdentifierList
from sqlparse.tokens import Keyword, Name

def extract_tables(sql):
    parsed = sqlparse.parse(sql)
    tables = set()
    
    for stmt in parsed:
        # 遍历所有
        for  in stmt.tokens:
            # 找到FROM/JOIN关键字
            if .ttype is Keyword and .value.upper() in ('FROM', 'JOIN'):
                # 下一个就是表名
                next_ = stmt._next(stmt._index())
                if next_:
                    tables.add(str(next_[1]))
    return tables

# 示例
sql = "SELECT a.*, b.col1 FROM table_a a JOIN table_b b ON a.id = b.id"
print(extract_tables(sql))
# 输出: {'table_a', 'table_b'}

当然，这只是最简版本。生产环境里你还要处理数据库限定名（db.table）、别名映射、以及各种方言差异。

3.2 解析存储过程获取字段级血缘

表级血缘只能告诉你「谁依赖谁」，但说不清「哪个字段依赖哪个字段」。要搞清楚字段级别的映射关系，就得啃存储过程了。

存储过程这东西，说白了就是一段带逻辑的SQL脚本。里面有变量声明、条件判断、循环、游标……解析起来比普通SQL复杂得多。

我常用的方法是：

• 变量追踪：把存储过程中的变量和表字段建立映射关系。比如v_name := (SELECT name FROM user WHERE id = 1)，那么v_name就依赖user.name。
• 赋值链分析：如果一个变量被多次赋值，要追踪最终来源。比如v_a := v_b; v_b := v_c; v_c := (SELECT col FROM t)，最终v_a依赖t.col。
• 条件分支处理：IF-ELSE分支里，同一个变量可能来自不同字段。这时候我建议取并集，宁可多报，不要漏报。

举个例子，下面这个简单的存储过程：

CREATE PROCEDURE proc_update_summary()
BEGIN
    DECLARE v_total DECIMAL(10,2);
    
    -- 从订单表汇总
    SELECT SUM(amount) INTO v_total FROM orders WHERE status = 'completed';
    
    -- 写入汇总表
    INSERT INTO daily_summary (total_amount, update_time)
    VALUES (v_total, NOW());
END;

解析后，我们能得到字段级血缘：daily_summary.total_amount ← orders.amount（经过SUM聚合）。

你想想看，如果没有这个字段级血缘，你只知道daily_summary依赖orders，但具体依赖哪个字段？不知道。一旦orders.amount改了，你根本不敢动。

3.3 利用日志解析ETL任务血缘

前面两种方式都是静态分析——看代码。但现实中有很多ETL任务，代码是黑盒的。比如你用DataX、Kettle、Informatica这些工具，生成的配置文件或者图形化任务，根本没法直接解析SQL。

怎么办？我的答案是：从日志入手。

ETL工具在执行时，通常会打印详细的日志，包括：

• 读取了哪些表、哪些字段
• 写入了哪些表、哪些字段
• 数据量、耗时、错误信息

这些日志就是血缘的「金矿」。

我总结了一套日志解析的流程：

步骤	操作	注意事项
1	日志采集	统一收集到ELK或S3，按任务ID和时间分区
2	日志解析	用正则或模板匹配，提取表名、字段名、操作类型
3	任务关联	同一个任务ID的读和写，构成一条血缘边
4	去重合并	同一任务多次执行，取最新的血缘关系

下面是一个日志解析的示例（假设日志格式是固定的）：

import re

log_pattern = re.compile(
    r'\[(?P<task_id>\w+)\]\s+'
    r'(?P<action>读取|写入)\s+'
    r'表:\s+(?P<table>\w+)'
)

def parse_etl_log(log_line):
    match = log_pattern.search(log_line)
    if match:
        return {
            'task_id': match.group('task_id'),
            'action': match.group('action'),
            'table': match.group('table')
        }
    return None

# 示例日志
logs = [
    "[TASK001] 读取 表: orders",
    "[TASK001] 写入 表: daily_summary",
    "[TASK002] 读取 表: users",
    "[TASK002] 写入 表: user_analysis"
]

for log in logs:
    result = parse_etl_log(log)
    if result:
        print(f"任务{result['task_id']} {result['action']}了{result['table']}")

日志解析的好处是：不依赖代码。无论你用的是什么ETL工具，只要日志格式统一，就能采集。坏处是：日志可能不完整。有些工具默认不打印字段级别的信息，那就只能拿到表级血缘。

三种采集方式的对比

我把这三种方式放在一起对比一下，方便你选型：

采集方式	血缘粒度	实现难度	适用场景	典型问题
SQL解析	表级	低	数据仓库、报表查询	方言差异、子查询嵌套
存储过程解析	字段级	高	核心业务、复杂加工	性能瓶颈、变量追踪
日志解析	表级/字段级	中	ETL工具、黑盒任务	日志格式不统一

我个人建议的落地策略是：先用SQL解析搞定80%的表级血缘，再用日志解析覆盖ETL任务，最后用存储过程解析攻坚核心链路。别想着一步到位，血缘采集是个持续迭代的过程。