4. 血缘采集技术栈：SQL解析、日志采集与API采集

血缘采集是数据血缘治理的起点，其技术栈覆盖了从静态代码解析到动态运行时捕获的完整链路。本章聚焦三大核心采集路径：SQL解析（Antlr、Calcite）、日志采集（Flume、Filebeat）以及API采集。理解各技术栈的适用场景、解析深度与集成成本，是构建可靠血缘系统的前提。

4.1 SQL解析：Antlr与Calcite

SQL解析是血缘采集的核心手段，适用于离线ETL脚本、数仓建模DDL以及BI报表查询。主流解析引擎包括Antlr和Apache Calcite，两者在解析能力、语法支持与集成方式上存在显著差异。

4.1.1 Antlr解析流程与血缘提取

Antlr通过语法文件（.g4）定义SQL文法，生成词法分析器与解析器。解析后得到AST，需通过自定义Visitor或Listener遍历节点，提取表名、字段名、别名及表达式中的引用关系。

// Antlr解析SQL并提取表名示例（伪代码）
CharStream input = CharStreams.fromString("SELECT a.id, b.name FROM table_a a JOIN table_b b ON a.id = b.id");
SqlLexer lexer = new SqlLexer(input);
CommonTokenStream tokens = new CommonTokenStream(lexer);
SqlParser parser = new SqlParser(tokens);
ParseTree tree = parser.statement();

// 自定义Visitor提取表引用
TableVisitor visitor = new TableVisitor();
visitor.visit(tree);
List<String> tables = visitor.getTables(); // ["table_a", "table_b"]

4.1.2 Calcite解析与血缘提取

Calcite将SQL解析为SqlNode，再通过SqlToRelConverter转换为RelNode（关系代数表达式）。血缘提取通常基于SqlNode的遍历，或直接利用Calcite的SqlValidator获取解析后的表与字段元数据。

// Calcite解析SQL并提取血缘关系（伪代码）
SqlParser parser = SqlParser.create("SELECT t1.col1, t2.col2 FROM tab1 t1 JOIN tab2 t2 ON t1.id = t2.id");
SqlNode sqlNode = parser.parseQuery();

// 使用SqlValidator获取解析后的元数据
SqlValidator validator = ...;
SqlNode validatedNode = validator.validate(sqlNode);

// 遍历SqlNode提取表与字段血缘
LineageExtractor extractor = new LineageExtractor(validator);
extractor.visit(validatedNode);
List<LineageRelation> relations = extractor.getLineage();

4.2 日志采集：Flume与Filebeat

日志采集适用于运行时血缘捕获，例如Hive/Spark执行计划日志、Presto/Trino查询审计日志、数据库Binlog等。Flume和Filebeat是两种主流的日志采集工具，分别面向不同的架构场景。

4.2.1 Flume血缘日志采集架构

Flume通过Source监听日志文件或网络端口，将日志事件写入Channel（内存/文件/Kafka），再由Sink写入HDFS、Kafka或Elasticsearch。在血缘场景中，通常使用Interceptor在Source端解析日志，提取血缘元数据。

# Flume配置示例：采集Hive执行日志并解析血缘
agent.sources = hiveSource
agent.sources.hiveSource.type = spooldir
agent.sources.hiveSource.spoolDir = /var/log/hive/exec
agent.sources.hiveSource.interceptors = lineageParser
agent.sources.hiveSource.interceptors.lineageParser.type = com.example.LineageInterceptor$Builder

agent.channels = kafkaChannel
agent.channels.kafkaChannel.type = org.apache.flume.channel.kafka.KafkaChannel
agent.channels.kafkaChannel.kafka.bootstrap.servers = kafka:9092
agent.channels.kafkaChannel.kafka.topic = hive-lineage

agent.sinks = kafkaSink
agent.sinks.kafkaSink.type = org.apache.flume.sink.kafka.KafkaSink
agent.sinks.kafkaSink.kafka.topic = hive-lineage
agent.sinks.kafkaSink.channel = kafkaChannel

4.2.2 Filebeat血缘日志采集架构

Filebeat以DaemonSet或Sidecar模式部署，监控日志文件变化，将日志行发送到Elasticsearch、Logstash或Kafka。在血缘场景中，Filebeat通常配合Logstash或自定义Processor进行日志解析与血缘提取。

# Filebeat配置示例：采集Trino审计日志
filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/trino/audit/*.log
  fields:
    log_type: trino-audit
  processors:
    - script:
        lang: javascript
        source: >
          function process(event) {
            var log = event.Get("message");
            // 解析审计日志，提取queryId、tables、columns
            var parsed = parseTrinoAudit(log);
            event.Put("query_id", parsed.queryId);
            event.Put("tables", parsed.tables);
            event.Put("columns", parsed.columns);
            return event;
          }

output.kafka:
  hosts: ["kafka:9092"]
  topic: "trino-lineage"
  partition.round_robin:
    reachable_only: false

4.3 API采集

API采集适用于外部系统血缘对接，例如从元数据平台（Atlas、DataHub）、数据开发平台（DolphinScheduler、Airflow）或数据服务层（GraphQL、RESTful API）获取血缘信息。API采集通常采用定时拉取或事件回调两种模式。

4.3.1 API采集接口设计示例

// 血缘API采集接口定义（RESTful）
GET /api/v1/lineage/tables?startTime=2025-01-01T00:00:00Z&endTime=2025-01-02T00:00:00Z

// 响应示例
{
  "code": 200,
  "data": [
    {
      "sourceTable": "ods.order",
      "targetTable": "dwd.order_detail",
      "operation": "INSERT",
      "fields": [
        {"sourceField": "order_id", "targetField": "order_id"},
        {"sourceField": "user_id", "targetField": "user_id"}
      ],
      "executionId": "exec_12345",
      "eventTime": "2025-01-01T12:30:00Z"
    }
  ],
  "nextOffset": "2025-01-01T12:30:00Z"
}

4.4 技术栈选型总结

血缘采集技术栈的选择需要综合考虑数据源类型、实时性要求、团队技术栈以及运维成本。以下表格总结了三大采集路径的推荐场景与关键考量。