📚 课程目录第3章流式数据血缘采集

3. 流式数据血缘采集：Flink SQL 的血缘解析、Kafka 消息的血缘追踪、实时数据流的字段级血缘

好，我们进入第三章。这一章讲的是流式数据的血缘采集。

说实话，流式数据的血缘比离线难搞得多。离线好歹有固定的表结构、固定的分区，你跑完一次调度就能把血缘算清楚。但流式数据是源源不断的，数据在 Kafka 里转瞬即逝，Flink 作业一重启状态就变了。你想想看，怎么追踪一条消息从 Kafka Topic A 经过 Flink 算子，最后落到 HBase 里的完整链路？

我刚开始做流批一体的时候，就被这个问题卡了很久。后来慢慢摸索出一套方法，今天分享给你。

3.1 Flink SQL 的血缘解析

Flink SQL 是流式计算的入口。大部分实时数仓的 ETL 逻辑，都是用 Flink SQL 写的。解析 Flink SQL 的血缘，说白了就是解析它的执行计划。

核心思路：Flink SQL 的 Calcite 优化器会生成一棵逻辑计划树。我们只需要遍历这棵树，就能提取出输入表、输出表、字段映射关系。

具体怎么做？我习惯用 Flink 的 TableEnvironment 来获取执行计划。代码很简单：

// 获取 Flink 执行计划
TableEnvironment tEnv = TableEnvironment.create(EnvironmentSettings.inStreamingMode());
tEnv.executeSql("CREATE TABLE source_table ...");
tEnv.executeSql("CREATE TABLE sink_table ...");
String explain = tEnv.explainSql("INSERT INTO sink_table SELECT ...");
System.out.println(explain);

这个 explain 输出里，包含了 Calcite 的 LogicalProject、LogicalFilter、LogicalTableScan 等节点。我们需要写一个解析器，把这些节点里的字段信息提取出来。

我在项目中遇到过一个问题：Flink 1.13 和 1.14 的执行计划格式不一样。1.13 用的是 JSON 格式，1.14 改成了 YAML。所以解析器要兼容两个版本。我的建议是统一用 Calcite 的 RelNode 接口来解析，不要直接解析字符串。

小技巧：Flink 1.15 之后，官方提供了 StatementSet 的 getJobGraph() 方法，可以直接拿到作业的血缘信息。如果你用的是新版本，优先用这个 API。

解析出来的血缘关系，我一般存成这样的结构：

字段	说明	示例
source_table	输入表名	ods_order
source_field	输入字段	user_id
sink_table	输出表名	dws_order_summary
sink_field	输出字段	uid
transform	转换逻辑	CAST(user_id AS BIGINT)

嗯，这里要注意：Flink SQL 里经常有 SELECT * 或者 GROUP BY 操作。对于 SELECT *，我们需要展开成所有字段；对于 GROUP BY，聚合字段和分组字段的血缘关系是不一样的。聚合字段的源头是原始字段经过函数计算，分组字段的源头就是原始字段本身。

3.2 Kafka 消息的血缘追踪

Flink SQL 的血缘解析只是第一步。真正难的是 Kafka 消息的血缘追踪。

为什么难？因为 Kafka 消息本身没有 schema。你往 Kafka 里写一条 JSON，它就是一个字符串。你没法直接从消息内容里知道它来自哪个表、经过了哪些算子。

我的做法是：在消息体里嵌入血缘标签。

具体来说，在 Flink 的 Sink 算子中，给每条消息加一个 _lineage 字段。这个字段记录了消息的来源：

{
  "user_id": 12345,
  "order_amount": 99.8,
  "_lineage": {
    "source_topic": "ods_order",
    "source_partition": 3,
    "source_offset": 1024,
    "operator_id": "cdc_source-1",
    "event_time": "2024-01-15T10:30:00Z"
  }
}

这样做的好处是：下游消费者拿到消息后，可以直接从 _lineage 字段里知道这条消息的完整来源。不需要去查元数据系统。

注意：嵌入血缘标签会增加消息体积。我建议只在关键链路（比如 ODS -> DWD）上开启这个功能，不要全链路都加。否则 Kafka 的存储和网络开销会暴涨。

我曾经在一个大项目里，全链路都加了血缘标签，结果每条消息从 200 字节膨胀到 500 字节。Kafka 的磁盘使用量直接翻了一倍多。后来改成只在关键节点加，才把成本降下来。

除了嵌入标签，还有一种做法是外部存储映射。就是把 Kafka 消息的 offset 和血缘关系存到外部系统（比如 HBase 或 Redis）里。消费者消费消息时，根据 offset 去查血缘关系。

两种方式各有优劣：

方式	优点	缺点
嵌入标签	消费者无需额外查询，实时性好	增加消息体积，修改 schema 麻烦
外部存储	不污染消息体，schema 灵活	消费者需要额外查询，有延迟

我个人习惯用嵌入标签的方式。因为实时排障的时候，你拿到一条消息就能直接看到它的来源，不用再去查别的系统。省事。

3.3 实时数据流的字段级血缘

字段级血缘，是血缘分析里最细的粒度。它要回答的问题是：输出结果的某个字段，到底来自输入数据的哪个字段？经过了什么计算？

在离线场景下，字段级血缘相对好做。因为 SQL 语句是静态的，解析一次就能拿到所有字段的映射关系。但在实时场景下，事情变得复杂了。

为什么？因为实时数据流里，字段的映射关系可能是动态的。比如：

使用了 CASE WHEN 条件分支，不同条件下字段来源不同
使用了 UDF（自定义函数），UDF 内部的字段映射是黑盒
使用了 JOIN 操作，字段可能来自左流或右流

我的做法是：在 Flink 算子层面做字段级血缘追踪。

具体来说，每个 Flink 算子（比如 MapFunction、FlatMapFunction）在运行时，会记录输入字段和输出字段的对应关系。这个关系可以是一个简单的映射表：

// 字段级血缘映射示例
Map<String, String> fieldLineage = new HashMap<>();
fieldLineage.put("output.user_id", "input.user_id");
fieldLineage.put("output.order_amount", "input.order_amount");
fieldLineage.put("output.total_amount", "input.order_amount * input.quantity");

然后，把这个映射关系上报到血缘系统。这样，当你在数据平台里看到某个字段时，可以一路追溯到它的源头。

关键点：字段级血缘的精度取决于算子的实现。如果算子内部逻辑复杂（比如多层嵌套的 UDF），字段级血缘可能只能做到「这个字段来自某个输入字段」，但无法精确到「经过了哪几步计算」。

我建议的做法是：先粗后细。先做到算子级别的血缘（知道数据经过了哪些算子），再逐步细化到字段级别。不要一开始就追求 100% 的字段级血缘，那样成本太高。

下面这张图展示了实时数据流中字段级血缘的采集流程：

最后，我想强调一点：字段级血缘不是万能的。它只能告诉你「这个字段来自哪里」，但不能告诉你「这个字段为什么是这个值」。后者需要结合数据质量监控和日志分析才能定位。

但有了字段级血缘，排障的效率能提升 80% 以上。以前你可能要花半天时间追查一个字段的来源，现在几分钟就能搞定。这就是血缘分析的价值所在。