第四章 批式数据血缘采集：从Spark SQL到离线ETL

批式场景下的血缘采集，说白了就是搞清楚「数据从哪来、经过什么变换、最终落到哪」。我做了这么多年离线数仓，发现很多团队在排障时最头疼的就是——一个报表数据对不上，得翻半天代码才能定位问题源头。如果血缘链路清晰，这种问题几分钟就能搞定。

这一章，我们重点聊三个核心场景：Spark SQL的血缘解析、Hive/数仓表的血缘构建、以及离线ETL的血缘链路。嗯，这三个场景基本覆盖了90%的批式数据流。

4.1 Spark SQL的血缘解析

Spark SQL是目前离线计算的主力引擎。它的血缘解析，我个人习惯从两个层面入手：逻辑计划层面和物理执行层面。

4.1.1 逻辑计划中的血缘信息

Spark SQL执行时，会先生成逻辑计划（Logical Plan）。这个计划里，天然就包含了输入表、输出表、字段映射关系。我们可以通过Spark的QueryExecution对象直接拿到。

举个例子，你执行一条SQL：

INSERT OVERWRITE TABLE dws.sales_summary
SELECT 
    region,
    product_id,
    SUM(amount) as total_amount
FROM dwd.sales_detail
WHERE dt = '2024-01-15'
GROUP BY region, product_id

Spark的逻辑计划会记录：

  • 输入：dwd.sales_detail（表）
  • 输出：dws.sales_summary（表）
  • 字段映射：region、product_id、SUM(amount)

我在项目中遇到过，有些同学直接解析SQL文本，用正则匹配表名。这种做法太脆弱了——一旦SQL里有子查询、CTE、或者用了USING语法，正则就很容易漏掉。我建议直接用Spark的QueryExecution.analyzed逻辑计划，它已经帮你把表名、字段都解析好了。

小技巧：使用 spark.sql("EXPLAIN EXTENDED " + yourSQL) 可以输出完整的逻辑计划文本，里面包含了输入输出表信息。但注意，这个输出是给人看的，机器解析还是建议用 QueryExecution API。

4.1.2 物理计划中的分区级血缘

逻辑计划只能告诉你「读了哪张表」，但实际排障时，我们往往需要知道「读了哪个分区」。比如上面那个SQL，如果dwd.sales_detail是分区表，我们想知道它到底读了dt='2024-01-15'这个分区。

这时候就要看物理计划了。Spark的物理计划里，FileSourceScanExec节点会记录实际读取的文件路径。通过解析文件路径，可以反推出分区信息。

我曾经踩过一个坑：某天凌晨的离线任务跑完，报表数据异常。查了半天，发现是上游表的分区字段dt类型从string改成了int，但SQL里还是用字符串比较。Spark在分区裁剪时，直接全表扫描了。如果当时有分区级的血缘监控，这个问题一眼就能发现。

4.2 Hive/数仓表的血缘构建

Hive表的血缘，比Spark SQL要复杂一些。因为Hive表可能被多种引擎访问：Spark、MapReduce、Tez、甚至直接通过Hive CLI。我们需要一个统一的血缘采集方案。

4.2.1 基于Hive Hook的采集方案

Hive提供了HiveMetastoreHook和HiveDriverRunHook两种扩展点。我个人推荐用HiveDriverRunHook，因为它能捕获到完整的SQL执行上下文。

基本思路是这样的：

  1. 实现HiveDriverRunHook接口
  2. 在preDriverRun方法中记录SQL开始执行
  3. 在postDriverRun方法中解析SQL，提取输入输出表
  4. 将血缘关系写入外部存储（比如Kafka或HBase）
核心代码片段：Hive Hook中解析SQL获取血缘 
public class LineageHook implements HiveDriverRunHook {
    @Override
    public void postDriverRun(HiveDriverRunHookContext hookContext) {
        String command = hookContext.getCommand();
        // 使用Hive的SemanticAnalyzer解析SQL
        ParseDriver pd = new ParseDriver();
        ASTNode tree = pd.parse(command);
        // 遍历AST，提取TOK_TABNAME和TOK_INSERT
        LineageInfo lineage = extractLineage(tree);
        // 写入血缘存储
        lineageStore.save(lineage);
    }
}

嗯，这里要注意：Hive的AST解析比较繁琐，建议直接用SemanticAnalyzer的getInputs()和getOutputs()方法，它已经帮你把输入输出表整理好了。

4.2.2 数仓分层表的血缘关系

数仓通常有ODS、DWD、DWS、ADS等分层。每一层之间的血缘关系，其实就是一个DAG（有向无环图）。我习惯用一张表来维护这种分层血缘：

字段 类型 说明
source_table string 源表（如 ods.order）
target_table string 目标表（如 dwd.order_detail）
etl_job string 执行ETL的任务名
field_mapping map<string,string> 字段级映射关系
execution_time timestamp 执行时间

有了这张表，你可以轻松回答「这张DWS表依赖哪些ODS表」或者「这个字段是从哪一层来的」。我在项目中，经常用这个表做影响分析——比如要修改某个ODS字段，先查一下会影响哪些下游表。

4.3 离线ETL的血缘链路

离线ETL的血缘，比单条SQL要复杂得多。一个完整的ETL任务，可能包含多个步骤：数据抽取、清洗、转换、聚合、加载。每个步骤都可能产生新的血缘关系。

4.3.1 多步骤ETL的血缘拼接

我见过最典型的场景是：一个Shell脚本里，调用了多个Spark SQL和Hive SQL。比如：

#!/bin/bash
# 步骤1：抽取ODS数据
spark-sql -f step1_extract.sql
# 步骤2：清洗DWD数据
spark-sql -f step2_clean.sql
# 步骤3：聚合DWS数据
spark-sql -f step3_aggregate.sql

这种场景下，血缘采集不能只看单条SQL。你需要把整个脚本的执行上下文串联起来。我的做法是：

  1. 在脚本开始前，生成一个全局的batch_id
  2. 每个SQL执行时，通过环境变量传递这个batch_id
  3. 血缘采集时，把batch_id和血缘关系一起存储
  4. 最后根据batch_id，把多个SQL的血缘拼接成一条完整链路
避坑指南：我曾经遇到过一个情况，ETL脚本里用了source命令加载了另一个脚本，导致血缘链路断掉了。后来我改成在脚本入口统一注册batch_id，并在所有子脚本中显式传递，才解决了这个问题。

4.3.2 依赖调度系统的血缘采集

大多数公司的离线ETL是通过调度系统（如Airflow、DolphinScheduler）来管理的。调度系统本身就有DAG依赖关系，我们可以利用这个信息来补充血缘。

举个例子，在Airflow中，每个DAG的task_id和upstream_task_ids天然就构成了任务级血缘。我们可以：

  • 从Airflow的元数据库读取DAG结构
  • 结合每个Task的SQL解析结果
  • 生成从「任务依赖」到「表依赖」的完整血缘图

我个人习惯把这种血缘分为三个层级：

  • 任务级：Task A → Task B（调度依赖）
  • 表级：Table X → Table Y（数据流动）
  • 字段级：Field a → Field b（字段映射）

三个层级组合起来，就是一个立体的血缘网络。排障时，你可以从任意一层切入，快速定位问题。

4.4 本章核心知识图谱

下面这张图，把批式血缘采集的核心链路画出来了。你可以看到，从Spark SQL、Hive SQL到离线ETL，血缘信息最终汇聚到统一的血缘存储中，供后续的排障和影响分析使用。

批式数据血缘采集核心链路 Spark SQL 任务 Hive SQL 任务 离线ETL 脚本 血缘采集器 Hive Hook / Spark Listener 调度系统元数据 血缘存储 (HBase / MySQL / Neo4j) 排障分析 / 影响分析 / 数据溯源 说明：批式血缘采集的核心是「多源采集 → 统一存储 → 上层应用」。Spark SQL通过QueryExecution解析，Hive通过Hook采集， 离线ETL则结合调度系统元数据和SQL解析，最终汇聚成完整的血缘链路。

这张图里，我特意把「调度系统元数据」也放进了采集层。为什么？因为很多离线ETL的血缘，光靠SQL解析是拿不全的——比如一个任务依赖另一个任务的输出，这种依赖关系只有调度系统知道。把调度元数据纳入血缘采集，才能形成完整的链路。

← 上一章 📖 返回目录 下一章 →