4、血缘采集技术：SQL解析、日志采集与手动标注

血缘采集，说白了就是搞清楚数据「从哪来、到哪去」。我做了这么多年数据治理，见过太多团队在血缘采集上栽跟头。有的全量解析跑不动，有的日志采集漏数据，还有的纯靠人工标注累到崩溃。

这一章，我把三种主流采集技术掰开揉碎讲清楚。你想想看，没有靠谱的血缘采集，后面的质量监控全是空中楼阁。

4.1 SQL解析：ANTLR与JSqlParser

SQL解析是血缘采集的硬核手段。它直接分析SQL语句的抽象语法树（AST），从中提取表、字段、函数之间的依赖关系。

我个人习惯把SQL解析分成两类：静态解析和动态解析。静态解析只分析SQL文本本身，动态解析还要结合元数据做上下文推断。

4.1.1 ANTLR：通用解析引擎

ANTLR（Another Tool for Language Recognition）是个强大的解析器生成器。它支持多种语法，包括SQL方言。我在项目中遇到过最头疼的场景——公司同时用了Hive SQL、Spark SQL和Trino，三种方言差异不小。用ANTLR，我只需要维护一份语法文件，就能生成对应的解析器。

核心思路：ANTLR通过.g4语法文件定义词法和语法规则，生成Lexer和Parser。解析后得到AST，再通过Visitor或Listener遍历树节点，提取血缘关系。

下面是一个简化的ANTLR解析流程示例：

// 1. 定义语法文件（简化版）
grammar SimpleSQL;

statement: SELECT column_list FROM table_ref where_clause? ;
column_list: column (',' column)* ;
column: IDENTIFIER ;
table_ref: IDENTIFIER ;
where_clause: 'WHERE' condition ;

// 2. 生成解析器
// 使用ANTLR工具：java -jar antlr-4.9.2-complete.jar SimpleSQL.g4

// 3. Java代码遍历AST
public class LineageExtractor extends SimpleSQLBaseVisitor<Void> {
    @Override
    public Void visitTable_ref(Table_refContext ctx) {
        String tableName = ctx.IDENTIFIER().getText();
        System.out.println("发现源表: " + tableName);
        return super.visitTable_ref(ctx);
    }
    
    @Override
    public Void visitColumn(ColumnContext ctx) {
        String colName = ctx.IDENTIFIER().getText();
        System.out.println("发现字段: " + colName);
        return super.visitColumn(ctx);
    }
}

我的经验：ANTLR适合处理复杂SQL和多种方言。但学习曲线陡峭，语法文件调试起来挺费时间。如果只是解析标准SQL，JSqlParser更轻量。

4.1.2 JSqlParser：轻量级SQL解析

JSqlParser是Java生态里最常用的SQL解析库。它直接解析SQL为Java对象，不需要生成代码。说白了，开箱即用。

我曾经在一个实时数据管道项目里用JSqlParser解析Spark SQL。每天几万条SQL，解析速度毫秒级，内存占用也低。嗯，这里要注意——JSqlParser对某些方言支持不够好，比如Hive的LATERAL VIEW EXPLODE，它解析会报错。

// JSqlParser解析示例
String sql = "SELECT a.user_id, b.order_amount " +
             "FROM dim_user a " +
             "JOIN fact_order b ON a.user_id = b.user_id " +
             "WHERE b.dt = '2024-01-01'";

CCJSqlParserManager parserManager = new CCJSqlParserManager();
Statement statement = parserManager.parse(new StringReader(sql));

if (statement instanceof Select) {
    Select select = (Select) statement;
    PlainSelect plainSelect = (PlainSelect) select.getSelectBody();
    
    // 提取源表
    List<Table> tables = plainSelect.getFromItem() instanceof Table ?
        Collections.singletonList((Table) plainSelect.getFromItem()) :
        Collections.emptyList();
    
    // 提取目标字段
    List<SelectItem> selectItems = plainSelect.getSelectItems();
    for (SelectItem item : selectItems) {
        if (item instanceof SelectExpressionItem) {
            SelectExpressionItem exprItem = (SelectExpressionItem) item;
            System.out.println("输出字段: " + exprItem.getAlias());
        }
    }
}
对比维度 ANTLR JSqlParser
学习成本 高（需学习.g4语法） 低（直接解析）
方言支持 强（可自定义语法） 中（标准SQL为主）
性能 中（生成代码后较快） 高（纯Java解析）
适用场景 多方言、复杂SQL 标准SQL、快速集成

4.2 日志采集：Spark Lineage与Airflow

SQL解析有个硬伤——它只能分析你写出来的SQL。但实际生产环境里，很多血缘关系藏在执行计划、调度日志和运行时上下文里。这时候，日志采集就派上用场了。

4.2.1 Spark Lineage：从执行计划提取血缘

Spark的DataFrame和Dataset API天然支持Lineage。每个DataFrame都维护了一个血统树（Lineage Tree），记录了它是从哪些数据源、经过哪些转换得来的。

我在一个实时数仓项目里用过Spark Lineage。当时需要追踪从Kafka到HBase的每条数据流。直接用df.explain(true)能看到完整执行计划，但解析起来挺麻烦。后来我写了个工具，从Spark的QueryExecution里提取Optimized Logical Plan，再解析出表级和字段级血缘。

关键API：Spark 2.4+提供了spark.listenerManager()注册自定义监听器，可以捕获每个SQL的执行计划。结合QueryExecution.optimizedPlan，能拿到最干净的血缘信息。

// 从Spark执行计划提取血缘
import org.apache.spark.sql.execution.QueryExecution
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan

def extractLineage(plan: LogicalPlan): List[LineageEdge] = {
    plan.collectLeaves().flatMap { leaf =>
        leaf match {
            case hdfsRelation: HdfsRelation =>
                // 提取HDFS文件路径对应的表
                val tableName = parsePathToTable(hdfsRelation.location)
                List(LineageEdge(tableName, hdfsRelation.output.map(_.name)))
            case _ => List.empty
        }
    }
}

// 注册Spark监听器
spark.sparkContext.addSparkListener(new SparkListener() {
    override def onExecutionStart(executionStart: SparkListenerExecutionStart): Unit = {
        val plan = executionStart.executionId
        // 这里可以拿到完整的执行计划
        println(s"Execution $plan started")
    }
})

避坑指南：我曾经踩过一个坑——Spark Lineage在DataFrame的某些转换操作（如withColumn）后，字段名会被自动重命名。解析时一定要处理ExpressionEncoder的字段映射，否则血缘会断掉。

4.2.2 Airflow：从调度日志提取血缘

Airflow本身不直接提供血缘采集功能。但它的DAG结构天然就是一张血缘图——每个Task的输入输出，就是数据流动的边。

我建议的做法是：在Airflow的TaskInstance回调里，注入自定义的Lineage Hook。每次Task执行完，把输入表、输出表、执行时间、运行状态写入血缘存储。

# Airflow Lineage Hook示例
from airflow.models import BaseOperator
from airflow.utils.decorators import apply_defaults

class LineageOperator(BaseOperator):
    @apply_defaults
    def __init__(self, source_tables, target_tables, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.source_tables = source_tables
        self.target_tables = target_tables
    
    def execute(self, context):
        # 执行实际业务逻辑
        result = self.do_business_logic()
        
        # 记录血缘
        lineage_record = {
            'dag_id': context['dag'].dag_id,
            'task_id': self.task_id,
            'execution_date': context['execution_date'],
            'source_tables': self.source_tables,
            'target_tables': self.target_tables,
            'status': 'success'
        }
        # 写入血缘存储（如Neo4j或Elasticsearch）
        write_lineage(lineage_record)
        return result

4.3 手动标注：兜底方案

说实话，自动采集再强，也总有覆盖不到的地方。比如：

  • 存储过程里的动态SQL
  • ETL工具（如Kettle、Informatica）的转换逻辑
  • 人工导入导出的Excel文件
  • 历史遗留系统的数据流转

这时候，手动标注就是最后的兜底方案。我见过有些团队完全依赖手动标注，结果几百张表标了半年还没标完。我的建议是：自动采集覆盖80%，手动标注补足20%。

4.3.1 手动标注的实践方法

手动标注不是让DBA去写SQL。我习惯用配置化标注的方式：

  1. 设计一个血缘标注的元数据表
  2. 开发一个简单的Web界面，让数据负责人拖拽连线
  3. 标注结果自动合并到血缘图谱中
-- 手动标注血缘的元数据表设计
CREATE TABLE manual_lineage (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    source_table VARCHAR(255) NOT NULL,
    source_field VARCHAR(255),
    target_table VARCHAR(255) NOT NULL,
    target_field VARCHAR(255),
    transform_expression TEXT,
    creator VARCHAR(100),
    create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    status TINYINT DEFAULT 1,  -- 1:有效 0:失效
    UNIQUE KEY uk_lineage (source_table, source_field, target_table, target_field)
);

-- 示例：标注一个字段映射
INSERT INTO manual_lineage 
(source_table, source_field, target_table, target_field, transform_expression, creator)
VALUES 
('ods_order', 'order_id', 'dwd_order_detail', 'order_no', 'CAST(order_id AS STRING)', 'zhangsan');

我的经验：手动标注一定要有版本管理。我曾经遇到有人误删了标注，结果整个血缘图乱了三天。后来加了valid_from和valid_to字段，支持时间回溯，再没出过问题。

4.4 三种技术的选型建议

你可能会问：这三种技术到底怎么选？我根据实际项目经验，给出一张选型决策表：

场景 推荐技术 理由
标准SQL、Hive/Spark SQL JSqlParser + ANTLR 解析准确，性能好
实时流处理（Spark Streaming） Spark Lineage 原生支持，无侵入
调度依赖复杂（Airflow） Airflow Hook 与调度系统深度集成
历史系统、存储过程 手动标注 兜底方案，灵活
混合架构（多种引擎） 组合使用 取长补短，覆盖全面

核心原则：自动采集为主，手动标注为辅。SQL解析覆盖结构化查询，日志采集覆盖运行时血缘，手动标注补足盲区。三者结合，才能构建完整的血缘图谱。

4.5 本章知识体系总览

下面这张SVG图，把三种血缘采集技术的核心逻辑和关系梳理清楚了。你可以把它当作本章的「知识地图」。

血缘采集技术知识体系 血缘采集技术知识体系 SQL解析 ANTLR JSqlParser 解析AST提取表/字段依赖 适合标准SQL和Hive/Spark SQL 静态分析，无运行时依赖 日志采集 Spark Lineage Airflow Hook 从执行计划/调度日志提取 运行时血缘，覆盖实时流 与计算引擎深度集成 手动标注 配置化标注 Web拖拽界面 兜底方案，补足盲区 适合存储过程/历史系统 需版本管理和审核机制 血缘图谱（Neo4j / Elasticsearch） 表级血缘 + 字段级血缘 + 转换逻辑 质量监控体系 血缘驱动数据质量规则 核心原则：自动采集覆盖80%，手动标注补足20%，三者结合构建完整血缘图谱

嗯，这一章的内容就到这里。三种技术各有优劣，关键是根据你的实际场景做组合。我个人习惯在项目中先用JSqlParser快速覆盖标准SQL，再用Spark Lineage补上实时流，最后留一个手动标注入口给运维团队。这样搭配下来，血缘覆盖率能做到95%以上。

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