Apache Atlas 与 Spark SQL 集成深度解析：捕获 LogicalPlan/PhysicalPlan 血缘的社区方案与生产实践

问题原文：Spark SQL 的执行计划（LogicalPlan/PhysicalPlan）能否被 Atlas 捕获？社区有哪些集成方案？

本文将系统性地回答上述问题。我们将以 ClickHouse 表自动分类治理 场景为背景，深入剖析 Spark SQL 执行计划（LogicalPlan/PhysicalPlan）被外部系统捕获的技术可行性，并全面评估当前社区主流的 Atlas 集成方案。文章将覆盖从 Spark Listener 机制、执行计划解析原理、到具体 Connector 的配置、验证及生产级调优的完整链路，旨在为拥有丰富 Spark 开发经验的工程师提供一条清晰、可落地的元数据集成路径。

一、场景引入：ClickHouse 表的血缘黑洞

在某实时分析平台，数据团队使用 Spark Structured Streaming 从 Kafka 消费原始日志，经过清洗、聚合后，写入 ClickHouse 供 BI 系统查询。核心作业如下：

// Spark 作业伪代码
val rawStream = spark.readStream.format("kafka").load()
val cleanedDf = rawStream.select(...)
val aggDf = cleanedDf.groupBy("user_id").agg(sum("clicks"))
aggDf.writeStream
  .format("clickhouse")
  .option("table", "dw.user_click_summary")
  .start()

业务痛点：

1. 血缘断裂：dw.user_click_summary 表在 Atlas 中存在，但其上游血缘为空。数据治理团队无法知道该表的数据来源于哪个 Kafka Topic。
2. 影响分析缺失：当 Kafka Topic raw_user_log 需要下线时，无法自动发现依赖它的 Spark 作业和 ClickHouse 表。
3. 合规风险：若原始日志中包含 PII 信息，无法追踪其是否流入了 ClickHouse。

核心诉求：能否让 Atlas 自动捕获 Spark 作业的执行计划，并建立从 Kafka Topic 到 ClickHouse 表的端到端血缘？

二、原理解析：Spark 执行计划捕获的可行性

1. Spark 的监听器（Listener）机制

Spark 提供了强大的事件监听机制，允许外部系统在作业生命周期的关键节点介入。与元数据捕获最相关的是 QueryExecutionListener。

• 触发时机：在每个 Spark SQL 查询（包括批处理和流处理）的 QueryExecution 对象被创建并完成优化后触发。
• 可访问对象：监听器可以获取到完整的 QueryExecution 实例，进而访问其 analyzed（逻辑计划）和 executedPlan（物理计划）。

生活化类比：QueryExecutionListener 就像汽车生产线上的“质检摄像头”。当一辆新车（Spark Query）在总装线（Catalyst Optimizer）上完成所有组装和调试后，在驶下生产线前，摄像头会拍下它的完整结构图（Logical/Physical Plan）。技术本质差异在于，这个“摄像头”是 Spark 主动提供的标准接口，任何符合规范的插件都可以接入，而非侵入式修改 Spark 内核。

2. 执行计划的结构与价值

• LogicalPlan：由 Catalyst 优化器生成的、与底层执行引擎无关的抽象语法树。它清晰地表达了数据的来源（如 Relation 节点）、转换（如 Project, Filter, Aggregate 节点）和去向。
• PhysicalPlan：针对特定执行引擎（如 Spark on YARN）优化后的、可执行的计划。它包含了分区、Shuffle 等细节。

对于血缘捕获，LogicalPlan 是黄金标准，因为它直接反映了用户 SQL 的语义，不包含执行层的噪音。

关键节点类型

节点类型	作用	血缘价值
DataSourceV2Relation	表示数据源（如 Kafka, JDBC, File）	输入血缘
Project	表示 SELECT 投影	字段映射
Aggregate	表示 GROUP BY 聚合	字段衍生
WriteToDataSourceV2	表示数据写出	输出血缘

通过遍历 LogicalPlan 树，理论上可以提取出完整的 inputs -> process -> outputs 三元组。

3. 为什么 Spark 官方没有内置 Atlas 集成？

Apache Spark 项目本身并未内置任何特定元数据系统的集成。这是出于以下设计哲学：

• 解耦：Spark 专注于计算，元数据管理应由专门的系统负责。
• 中立性：避免与某个商业或开源元数据产品绑定。
• 灵活性：允许用户根据自身需求选择不同的集成方案。

因此，所有 Spark 与 Atlas 的集成都依赖于社区驱动的第三方库。

三、社区主流集成方案深度对比

目前，社区主要有两大类方案：基于 QueryExecutionListener 的通用方案 和 针对特定数据源的定制方案。

1. Hortonworks Spark Atlas Connector (hortonworks-spark/spark-atlas-connector)

这是最广为人知的方案，由 Hortonworks（现 Cloudera）贡献。

工作原理

1. 注册 Listener：通过 --packages 引入 jar 包后，自动向 SparkSession 注册 AtlasSparkAgent。
2. 解析 LogicalPlan：在 onSuccess 回调中，调用自研的 SparkLineageReporter 类，递归遍历 LogicalPlan。
3. 构建 Entity：将识别出的数据源、目标和转换过程，构建成 Atlas 的 spark_process、kafka_topic、hive_table 等实体。
4. 上报：通过 Atlas Java Client 同步或异步上报。

优点

• 开箱即用：对常见的数据源（Hive, Kafka, HDFS）有良好支持。
• 社区成熟：经过多年生产环境验证。

缺点

• 字段级血缘弱：主要聚焦于表/Topic 级别，对复杂投影和聚合的字段级血缘支持有限。
• 维护停滞：随着 Hortonworks 被收购，该项目更新缓慢，对 Spark 3.x+ 的新特性（如 DataSource V2）支持不完善。
• 侵入性强：需要在每个 Spark 作业的提交命令中显式添加 --packages。

配置示例

# 提交 Spark 作业
spark-submit \
  --packages com.hortonworks:spark-atlas-connector_2.12:0.1.0-SNAPSHOT \
  --conf spark.sql.queryExecutionListeners=com.hortonworks.spark.atlas.SparkAtlasEventTracker \
  your_application.jar

2. 自研 Spark Listener + Atlas Java Client

对于有较强研发能力的团队，自研是更灵活、可控的选择。

核心步骤

1. 实现 QueryExecutionListener
2. 编写 Plan Parser：利用 Spark 的 TreeNode API 遍历计划树。
3. 集成 Atlas Client：使用 org.apache.atlas.AtlasClientV2 上报实体。

自研代码片段

// 自定义 Spark Listener
public class AtlasLineageListener implements QueryExecutionListener {
    private final AtlasClientV2 atlasClient;

    public AtlasLineageListener() {
        // 从配置中初始化 Atlas Client
        this.atlasClient = new AtlasClientV2(new String[]{"http://atlas-server:21000"}, "admin", "admin");
    }

    @Override
    public void onSuccess(String funcName, org.apache.spark.sql.execution.QueryExecution qe, long durationNs) {
        try {
            // 1. 获取优化后的逻辑计划
            LogicalPlan logicalPlan = qe.analyzed();
            
            // 2. 解析血缘
            LineageInfo lineage = parseLineage(logicalPlan);
            
            // 3. 构建 Atlas Entity
            List<Referenceable> entities = buildAtlasEntities(lineage);
            
            // 4. 上报
            EntityMutationResponse response = atlasClient.createEntities(entities);
            LOG.info("Successfully reported {} entities to Atlas", entities.size());
        } catch (Exception e) {
            LOG.error("Failed to report lineage for query: {}", funcName, e);
        }
    }

    // 核心解析方法（简化版）
    private LineageInfo parseLineage(LogicalPlan plan) {
        LineageInfo info = new LineageInfo();
        plan.foreach(node -> {
            if (node instanceof DataSourceV2Relation) {
                // 处理输入源
                DataSourceV2Relation source = (DataSourceV2Relation) node;
                info.addInput(extractSourceFromRelation(source));
            } else if (node instanceof WriteToDataSourceV2) {
                // 处理输出目标
                WriteToDataSourceV2 sink = (WriteToDataSourceV2) node;
                info.addOutput(extractSinkFromWrite(sink));
            }
            return null;
        });
        return info;
    }
}

⚠️ 警告：自研方案必须处理好异常隔离。Listener 中的任何未捕获异常都会导致整个 Spark 作业失败。务必使用 try-catch 并记录详细日志。

3. 方案对比总结

特性	Hortonworks Connector	自研 Listener	Hive Hook (作为参照)
开发成本	低	高	无（官方内置）
字段级血缘	弱	强（可控）	极强
数据源支持	有限（Hive/Kafka/HDFS）	任意（可扩展）	仅 Hive
维护性	社区停滞	自主可控	官方维护
侵入性	需改提交脚本	需打包进应用	仅需配 Hive
适用场景	快速验证、简单血缘	复杂血缘、生产核心	Hive 生态

四、生产级实践：以 ClickHouse 血缘捕获为例

1. 目标

捕获如下 Spark 作业的血缘：

• 输入：Kafka Topic raw_user_log
• 输出：ClickHouse 表 dw.user_click_summary
• 过程：user_id, SUM(clicks)

2. 自研方案实施步骤

步骤 1: 定义 ClickHouse 元模型

首先，需要在 Atlas 中为 ClickHouse 创建自定义 Type。

// clickhouse_model.json
{
  "enumDefs": [],
  "structDefs": [],
  "classificationDefs": [],
  "entityDefs": [
    {
      "superTypes": ["DataSet"],
      "name": "clickhouse_table",
      "description": "ClickHouse table metadata",
      "typeVersion": "1.0",
      "attributeDefs": [
        {"name": "name", "typeName": "string", "isOptional": false},
        {"name": "db", "typeName": "string", "isOptional": false},
        {"name": "columns", "typeName": "array<clickhouse_column>", "isOptional": true},
        {"name": "cluster", "typeName": "string", "isOptional": false}
      ]
    },
    {
      "superTypes": [],
      "name": "clickhouse_column",
      "description": "ClickHouse column metadata",
      "typeVersion": "1.0",
      "attributeDefs": [
        {"name": "name", "typeName": "string", "isOptional": false},
        {"name": "dataType", "typeName": "string", "isOptional": false},
        {"name": "comment", "typeName": "string", "isOptional": true}
      ]
    }
  ]
}

注册模型：

curl -u admin:admin -X POST -H "Content-Type: application/json" \
-d @clickhouse_model.json http://atlas:21000/api/atlas/v2/types/typedefs

步骤 2: 增强 Plan Parser

在自研 Listener 的 parseLineage 方法中，增加对 ClickHouse Sink 的识别。

// 在 parseLineage 方法中
else if (node instanceof WriteToDataSourceV2) {
    WriteToDataSourceV2 writeNode = (WriteToDataSourceV2) node;
    DataSourceWriter writer = writeNode.writer();
    if (writer instanceof ClickHouseWriter) { // 假设存在此类
        String tableName = extractClickHouseTableName(writer);
        info.addOutput(new DataAsset("clickhouse_table", tableName, "dw", "prod_cluster"));
    }
}

步骤 3: 验证血缘上报

1. 运行 Spark 作业
2. 检查 Atlas Server 日志

   grep "AtlasLineageListener" /var/log/atlas/application.log
   

   验证点：应看到成功上报的日志。
3. 通过 REST API 查询血缘

   # 获取 ClickHouse 表 GUID
   CH_GUID=$(curl -s -u admin:admin "http://atlas:21000/api/atlas/v2/entity/uniqueAttribute/type/clickhouse_table?attr:qualifiedName=dw.user_click_summary@prod_cluster" | jq -r '.entity.guid')
   
   # 查询上游
   curl -u admin:admin "http://atlas:21000/api/atlas/v2/lineage/upstream?guid=$CH_GUID"
   

   验证点：返回结果中应包含 kafka_topic 类型的实体，其 qualifiedName 为 raw_user_log@kafka-cluster。

3. 性能与可靠性调优

• 异步上报：不要在 Listener 的主线程中同步调用 Atlas API。应将 Entity 放入队列，由后台线程池处理，避免阻塞 Spark 作业。
• 批量发送：累积一定数量的 Entity 后再批量上报，减少网络开销。
• 重试与降级：对 Atlas API 调用增加指数退避重试机制。如果 Atlas 服务不可用，应将 Entity 暂存到本地磁盘或 Kafka，待恢复后重放。
• 资源隔离：为 Listener 的后台线程池分配独立的 CPU 和内存资源，防止与 Spark 计算任务争抢。

五、FAQ 与未来演进

Q1: Spark Structured Streaming 的血缘如何捕获？

A1: 可以捕获，但有挑战。每个微批次（micro-batch）都会触发一次 QueryExecutionListener。需要在 Listener 中识别出这是流作业，并将多次上报的血缘合并为一个逻辑上的持续血缘。同时，要注意避免因微批次过多而导致 Atlas 写入压力过大。

Q2: 能否捕获 DataFrame API 的血缘，而不仅仅是 SQL？

A2: 完全可以。无论是 spark.sql("SELECT ...") 还是 df.select(...).groupBy(...)，最终都会被 Catalyst 转换为相同的 LogicalPlan。Listener 捕获的是底层的 Plan，与上层 API 无关。

Q3: 与 OpenLineage/OpenMetadata 的 Spark Agent 相比如何？

A3: OpenLineage 是更现代的标准。它定义了一套与具体元数据系统解耦的血缘数据模型（Run, Job, Dataset）。Spark 的 OpenLineage Agent 会将血缘发送到 Marquez 或 OpenMetadata。如果你的公司采用 OpenMetadata，这可能是比 Atlas 更好的选择。但如果你已重度投资 Atlas，自研 Listener 仍是最佳路径。

Q4: Spark 3.3+ 的 DataSource V2 对血缘捕获有何影响？

A4: 利大于弊。DataSource V2 提供了更清晰、更一致的 API 来描述数据源和接收器。DataSourceV2Relation 和 WriteToDataSourceV2 节点使得在 Plan 中识别 I/O 操作变得更加容易和可靠，降低了自研 Parser 的复杂度。

Q5: 如何监控 Spark 血缘上报的健康度？

A5: 建议监控以下维度：

• Listener 错误率：通过 Spark 应用日志中的 ERROR 日志统计。
• Entity 上报延迟：在 Entity 中添加时间戳，对比其与 Spark 作业完成时间的差值。
• Atlas 写入成功率：监控 Atlas Server 的 entity-created JMX 指标。

未来演进方向

1. 标准化：推动 Spark 社区采纳类似 OpenLineage 的标准血缘接口。
2. 云原生集成：与 Kubernetes Operator 结合，自动为 SparkApplication CRD 注入血缘上报 Sidecar。
3. AI 辅助解析：利用 LLM 对复杂的 UDF 或 SQL 片段进行语义分析，提升字段级血缘的准确率。

总结

Spark SQL 的执行计划（LogicalPlan/PhysicalPlan）完全能够被外部系统捕获，其技术基础是 Spark 提供的 QueryExecutionListener 机制。然而，Apache Atlas 并未提供官方的、开箱即用的 Spark 集成方案。社区主流的 Hortonworks Connector 方案存在维护停滞、功能局限等问题。对于追求高可靠性和字段级血缘精度的生产环境，自研基于 QueryExecutionListener 的血缘上报模块，并结合 Atlas Java Client，是当前最可行、最灵活的解决方案。通过本文提供的 ClickHouse 血缘捕获案例，具备深厚 Spark 开发经验的工程师可以快速构建起适用于自身业务场景的元数据管道，从而打通 Spark 作业的血缘黑洞。

作者署名：九师兄

• 专题目录：【Apache Atlas】Apache Atlas 资深工程师到专家实战之路目录
• 总目录：【目录】技术体系目录

注意：本文由 AI 辅助生成，技术细节请以官方文档为准。生产环境使用前务必充分测试。