📚 课程目录第3章血缘采集机制与生产接入

3. 血缘采集机制与生产接入

血缘采集的本质，是把计算引擎“已经理解过的语义”重新翻译成一张表与表、字段与字段之间的有向图。不同引擎暴露语义的时机和位置不同：SQL 引擎在语法树与逻辑计划阶段最完整，Spark 在逻辑计划与物理计划之间最稳定，Flink 则在流式作业的算子拓扑与状态描述中最清晰。因此采集方案不能只做一种，而要按引擎选择“挂载点”。

一、三类引擎的血缘提取原理

SQL 引擎（Hive、Spark SQL、Presto/Trino）：核心是解析 AST 并做语义分析。以 INSERT INTO dw.dws_order SELECT ... FROM ods.order JOIN ods.user ... 为例，解析器先产出抽象语法树，分析器再绑定 Catalog 元数据，把 ods.order、ods.user 解析为真实表，把 SELECT 中的表达式解析为列级依赖。列级血缘的关键在于：对每个输出列，沿表达式树回溯到源列，遇到 JOIN 条件、WHERE 过滤、聚合函数时分别记录“关联依赖”“过滤依赖”“聚合依赖”。

Spark：血缘信息藏在 LogicalPlan 与 QueryExecution 中。通过 spark.sql(...).queryExecution.analyzed 或监听 QueryExecutionListener，可以拿到 UnresolvedRelation、Project、Aggregate、Join 等节点。相比 SQL 文本解析，逻辑计划已经完成了列绑定，能直接给出 outputSet 与 inputSet 的映射，因此列级血缘更准确，尤其适合 DataFrame/Dataset API 场景。

Flink：流式作业的血缘体现在 StreamGraph 到 JobGraph 的转换中。每个算子（Source、Map、KeyBy、Window、Sink）携带输入输出算子 ID 与表标识。通过 StreamExecutionEnvironment.getStreamGraph() 或 SQL Gateway 的 Operation 解析结果，可以还原“源表 → 中间算子 → 结果表”的链路。Flink 的难点在于算子链融合（Operator Chaining）会隐藏中间节点，需要在采集时按算子 UID 展开。

二、低侵入采集方案设计

生产环境最忌讳“为了血缘改业务代码”。低侵入的核心思路是：不改 SQL、不改作业逻辑，只在引擎的扩展点挂监听。具体分三层落地。

第一层：SQL 网关代理。在 HiveServer2、Trino Gateway、Spark Thrift Server 前挂一层代理，拦截 ExecuteStatement 请求，把 SQL 文本旁路到解析服务。解析服务用 Calcite 或引擎自带 Parser 生成 AST，再结合元数据服务补全表名。这种方式对业务完全透明，缺点是拿不到执行后的真实计划，遇到动态 SQL 会失效。

第二层：引擎事件监听。Spark 用 QueryExecutionListener，Flink 用 ExecutionListener，Hive 用 HiveHook（PreExecute/PostExecute）。以 Spark 为例：

spark.listenerManager.register(new QueryExecutionListener {
  override def onSuccess(funcName: String, qe: QueryExecution, duration: Long): Unit = {
    val plan = qe.analyzed
    val relations = plan.collect { case r: UnresolvedRelation => r.tableName }
    val lineage = extractColumnLineage(plan)  // 自定义列级提取
    lineageSink.send(lineage)
  }
  override def onFailure(funcName: String, qe: QueryExecution, ex: Exception): Unit = {}
})

这种方式能拿到已绑定的逻辑计划，列级血缘准确率高，且对业务代码零侵入。注意监听器要异步发送、批量聚合，避免阻塞主作业。

第三层：作业提交侧埋点。在 YARN/K8s 提交脚本中注入环境变量，把作业 ID、提交人、队列、引擎版本一并上报，用于血缘图上的“作业归属”标注。这一层不解析语义，只补上下文。

三、动态 SQL 与 UDF 的血缘断裂处理

动态 SQL 的典型形态是字符串拼接：

val dt = "2024-06-01"
val sql = s"""
  INSERT OVERWRITE TABLE dws_order_dt PARTITION(dt='$dt')
  SELECT o.order_id, u.user_name, o.amount
  FROM ods.order o JOIN ods.user u ON o.uid = u.uid
  WHERE o.dt = '$dt'
"""
spark.sql(sql)

静态解析器看到的是 spark.sql(sql)，无法还原真实表名。处理策略是“运行时捕获 + 模板回填”：在监听器中拿到 analyzed 计划后，直接读取 LogicalRelation 中的 catalogTable，此时表名、分区、列都已确定，血缘自然完整。对于拼接前的模板，可在提交侧记录 SQL 模板与变量绑定关系，作为血缘的“可追溯注释”。

UDF 是更棘手的黑盒。以 SELECT my_hash(phone) AS phone_hash FROM ods.user 为例，解析器只能看到 my_hash 是一个 ScalaUDF，无法知道它内部读了哪些列。处理方式有三类：

注册时声明依赖：在 UDF 注册接口中增加 inputColumns 与 outputColumns 元数据，采集时读取该声明，把 UDF 节点展开为“输入列 → UDF → 输出列”的子图。
保守标记：无法声明时，把 UDF 的输出列标记为“依赖该 UDF 全部输入列”，并在血缘图上用虚线边表示“不确定依赖”，供下游人工确认。
字节码/表达式分析：对简单 UDF（如仅做字符串拼接、类型转换），可通过分析其表达式树或字节码提取列引用；复杂 UDF 不建议强行推断，避免产生错误血缘。

实践建议：把 UDF 分为“可声明”“可分析”“黑盒”三档，分别对应精确血缘、近似血缘、标记血缘。血缘图上的边用不同线型区分，下游影响分析时对“标记血缘”给出告警而非直接阻断。

四、采集结果的一致性校验

采集上线后，必须回答“血缘对不对”。可用一组已知答案的作业做回归：准备 10 个覆盖 JOIN、聚合、窗口、UDF、动态分区的 SQL，人工标注期望的表级与列级血缘，每次采集版本升级后自动比对。若列级血缘的准确率低于 95%，优先排查 UDF 展开与动态 SQL 捕获两个环节。同时，对同一张表被多个作业写入的情况，检查血缘图是否存在环——正常数仓 DAG 不应有环，出现环通常意味着临时表命名冲突或解析错误。

最后，采集服务本身要可观测：记录每次解析的耗时、失败原因、血缘边数量，按引擎和作业类型分桶统计。当某个引擎的解析失败率突增时，往往意味着引擎版本升级导致挂载点 API 变化，需要及时适配。