Spark native向量化组件初探：用 DataFusion Comet 给 Spark 提速
说说我最近一段时间在折腾的一件事：给一套老旧的 Spark 作业换上 DataFusion Comet 这套 native 向量化组件。先说结论，原来跑一个多小时的大查询，现在基本能压进四十分钟以内，有些简单扫描场景甚至能跑到两倍左右的提升。这篇文章不打算做概念复读，就把我在选型、编译、配置、排障过程中真正踩过的坑和看到的收益整理出来，希望对正在考虑要不要引入这套方案的同学有些实际帮助。
这套组件解决的痛点非常直接：Spark 作业明明给了几十个 executor，但 CPU 使用率经常上不去，大量时间花在 JVM 的序列化、GC 和逐行处理上。DataFusion Comet 的思路，是用一套基于 Arrow 列式内存格式的 native 执行引擎，把查询执行的一部分操作直接搬到 JVM 之外，绕开传统行式处理的额外开销。适合的人群也很明确：已经在用 Spark SQL 做离线数仓处理，觉得性能上不去，又不愿意把整个架构推倒重写的团队。对底层原理了解得越多，排查问题就越快，这也是我把原理部分排在实操前面的原因。
1. 先搞清楚 Comet 到底在解决什么问题
1.1 Spark 执行引擎的“慢”到底慢在哪
很多人在评估大数据组件时只盯着 benchmark 数字，却忽略了性能瓶颈的来源。想理解 DataFusion Comet 为什么能提速，必须先把 Spark 执行引擎的几个天然短板拆开看。
第一，JVM 行式处理的天花板。Spark 的物理执行计划最终会被拆成一个个 Task，这些 Task 在 JVM 里以行迭代的方式处理数据。虽然 Spark 有 WholeStageCodegen 这种把多个算子编译成一段 Java 代码的技巧，但它仍然是行式模型。每行数据都要经过虚方法调用、类型判断和对象创建，CPU 的指令流水线很难跑满。你可以把它想象成一条流水线上每个工位只处理一件产品，传送带却要反复启停，吞吐量注定上不去。
第二，序列化和反序列化的开销被低估。Shuffle 和 Join 是 Spark 作业里最昂贵的环节，因为数据必须经过序列化写入磁盘或网络，到下游再反序列化。JVM 的序列化机制走的是反射加字节码生成，虽然 Kryo 比 Java 原生序列化快不少，但依然逃不掉逐对象处理。当数据量达到 TB 级别时，这部分耗时占比相当可观。
第三，GC 带来的不确定延迟。Executor 内存里全是 Java 对象，大 shuffle 之后马上就会有一轮 Young GC 甚至 Full GC。GC 停顿会让整个 Task 的执行时间出现长尾，在实时性要求高的场景里很难受。代码生成产生的类对象越多，Metaspace 压力也越大。
第四，向量化能力缺失。现代 CPU 的 SIMD 指令可以一次对多个数据执行同一操作，比如一次比较 8 个浮点数。但 JVM 的行式处理很难有效利用 SIMD，因为数据在内存里不是连续排列的，每次都要 gather 数据。列式存储天然适合 SIMD，但 Spark 默认的 Tungsten 格式在这方面做得并不彻底。
1.2 DataFusion Comet 的定位和设计目标
DataFusion Comet 选了一个非常聪明的切入点：不是另起炉灶做一个新的计算引擎，而是作为 Spark 插件存在。它的设计目标是在不改变 Spark 使用方式的前提下，把核心执行路径的一部分替换成 native 向量化执行。
具体怎么理解？你可以把 Spark 看成一家物流公司，Driver 还是那个调度中心，负责接收订单、规划路线、拆分任务。但到了具体搬运环节，原来派出去的是 JVM 工人，现在换了一批更高效的分拣机器，这批机器就是 DataFusion 引擎。订单还是照常下，客户感觉不到调度方式变了，但分拣速度明显变快。
这套组件的核心模块是原生执行引擎，基于 Rust 编写的 DataFusion 实现。Rust 的优势在于无 GC、内存安全且能做紧凑的列式内存布局，再加上 Arrow 列式内存格式，可以非常自然地做向量化计算。Comet 接受 Spark 传来的物理执行计划，把它转换成一个 native 执行计划，用 JNI 调用的方式拉到 JVM 外面跑。
它的另一个巧妙之处是回退机制。并非所有操作都能被 native 引擎接管，遇到不支持的操作，Comet 会把该部分执行计划交还给 Spark 自己执行。这种设计保证了稳定性和正确性，不会说因为遇到一个不支持的算子就整个作业失败。这也是我认为它适合生产环境的最重要原因：渐进式替换，而不是非黑即白。
2. 核心机制：从 Spark 算子到 native 代码的完整链路
2.1 Shim 层：多版本兼容的基石
DataFusion Comet 有一个自己定义的 Spark 版本适配层，业界一般叫 Shim 层。为什么要做这么一层？因为 Spark 的源码接口在不同版本之间变动频繁，比如内部执行计划节点的定义、内存管理的接口、Shuffle 的读写方式，甚至一些工具类的包路径都会变。如果直接依赖某个特定版本的 Spark 接口，插件就只能在这个版本上用，适用范围太窄。
Comet 的设计思路是把与 Spark 版本相关的代码抽出来，通过抽象接口隔离。每个支持的 Spark 版本实现一套对应的 Shim，Comet 主代码走的是统一接口。这样做的好处不只是兼容性好，更重要的是主逻辑可以保持相对稳定，修复核心 bug 不用每个版本都做一遍。
我在实际编译时发现，不同 Spark 版本对应的 artifactId 后缀也不一样，比如 Spark 3.3、3.4、3.5 分别对应不同的 profile。所以编译前一定要确认自己的 Spark 主版本，再选择对应的 profile 参数。这一点我后面在实操部分会再展开。
2.2 JNI 桥与 native 算子树
整个执行链路可以这样描述：Spark Driver 生成逻辑计划和物理计划之后，Comet 的规则会把能用 native 引擎执行的物理算子标记出来。这些算子包括扫描、过滤、投影、聚合、排序、Join、Shuffle 等。然后每个 executor 上的 JVM 侧组件通过 JNI 调用 native 库，把执行计划传给 Rust 侧的 DataFusion 执行器。
这里需要提一下 Arrow 列式内存格式的作用。数据从 Spark 的 InternalRow 转为 Arrow RecordBatch 后，在内存里的布局就是连续排列的列数组。每个列的类型是确定的，比如 Int64 列在内存里就是连续的一串 64 位整数。这种布局有几个直接好处：CPU 缓存命中率高，因为扫描的时候读的是连续内存；SIMD 指令可以直接作用在数组上；数据可以零拷贝地在不同算子之间传递，减少内存拷贝。
JNI 部分有一个容易被忽视的问题：跨语言传对象不能直接传引用，需要做序列化或内存拷贝。Comet 的做法是尽量用 Arrow 格式直接传递内存地址和长度信息，避免逐字段拷贝。数据量大的时候，这种设计能节省大量时间。但 JNI 调用本身是有开销的，Comet 的批次设计会尽量一次处理较大的数据块，把 JNI 调用次数压缩下来。
2.3 算子支持范围与回退策略
Comet 目前支持的算子覆盖了绝大多数常规 SQL 场景。我把常见算子列一下：Table Scan、Filter、Project、HashAggregate、Sort、HashJoin、SortMergeJoin、Shuffle 的写入和读取、Window 函数的一部分场景。对于不支持的情况，比如某些复杂表达式、自定义 UDF 里访问了特定对象、部分子查询，Comet 会整体回退给 Spark 执行。
回退策略非常重要。我看到一些团队对引擎替换抱着“只要有一个不支持的场景就全盘放弃”的态度，这其实是误解。Comet 的执行计划会被拆成片段，支持的部分走 native，不支持的部分保留 Spark。只要整体的执行计划是合理的，就能拿到大部分收益。
不过回退也带来一个性能陷阱：如果一个 Fragment 里同时包含 native 和 Spark 算子，中间要经历 Arrow 格式和行格式的转换，这部分开销不容小觑。所以我做性能调优时，会先通过 EXPLAIN 看一下执行计划里是否频繁出现转换点。如果转换太频繁，考虑调整参数让更多操作被划分到同一边。
3. 环境准备与接入手把手实操
3.1 版本选型与前置依赖
先概括一下我当时选型时的考量。Comet 对 Spark 版本的要求比较明确，当时我测试环境是 Spark 3.4，就选择匹配 Spark 3.4 的 profile。因为 Jar 包是和 Spark 版本绑定的，版本不匹配在运行时大概率会报序列化或类转换的异常，非常难排查，这是整个接入过程中最容易踩的坑。
系统层面的依赖包括 JDK 8 或 11、Maven 3.6+、Rust 工具链。Rust 是编译 native 库必需的，虽然主要的组件官方文档建议用特定版本，但一般只要 Rust 版本不太老，问题都不大。编译过程中还需要 protoc，用来处理 Arrow 的 Protocol Buffers 定义。LLVM 是否必需取决于目标平台，在 Linux 环境编译时，可能需要确认系统里有 clang 相关的库。
还有一个容易忽略的点：native 库编译完成后，运行时对 glibc 版本有要求。如果你的集群机器操作系统比较老，glibc 版本过低会导致动态库加载失败。建议在部署前先到 executor 机器上做一个最小化的冒烟测试，直接写一个简单的 DataFrame 查询，看能不能正常跑通。
3.2 编译打包过程记录
我当时在编译服务器上执行的操作大致是这样：先确认环境变量，然后克隆 DataFusion Comet 项目源码，切换到与目标 Spark 版本对应的分支或 tag，这一步很重要，直接决定了后面代码能否编译通过。接下来用 Maven 命令触发编译，关键参数是 profile 选择，Spark 3.4 对应-Pspark-3.4。整个编译过程会把 Rust 的 native 代码也一起构建出来，所以耗时比较长，二十分钟到一个小时都正常。
编译完成后，需要关注的产品物主要有两个：一个是可以直接放到 Spark 的 jars 目录里的 jar 包，另一个是 native 动态库文件，比如 libcomet.so。如果你是用 spark-submit 的方式提交作业，可以通过--jars把 jar 带上，同时通过--conf指定 native 库的加载路径，比如spark.comet.native.lib.path。如果是在集群模式下部署，建议把动态库放到所有 executor 节点都能访问到的地方，我一般放在/opt/comet/lib下面，然后在 Spark 配置里统一指向。
3.3 Spark 配置与启停验证
编译好以后，配置是下一个关键环节。最基础的开关有三个：spark.comet.enabled决定是否启用 Comet 整体开关；spark.comet.exec.enabled决定是否启用 native 执行引擎替换 Spark 算子；spark.comet.exec.mode控制执行模式，auto 模式对支持的操作走 native，对不支持的操作回退 Spark，兼容性最高，建议第一轮先用这个模式。
我还调整了几个内存相关参数。Comet 有自己的 native 内存分配机制，和 Spark 原来的执行内存是分开的。spark.comet.memory.offheap.enabled开启之后，Comet 会尽量在堆外分配内存，这样和 Arrow 的零拷贝机制配合更好。但是要注意spark.memory.offHeap.size是 Spark 的堆外总内存限制，如果开得不够，可能出现内存不足。我当时是给每个 executor 设了 8G 的堆外内存，再配合设置 COMET 页缓存大小。
配置完成后，验证是否生效有个简单方法：在 Spark UI 的 SQL 页签里查看物理执行计划，如果执行计划里出现CometScan、CometFilter、CometHashAggregate之类的节点，说明对应部分已经被 native 引擎接管了。我还会做一个 EXPLAIN 输出，确认关键算子前缀是Comet，这样能快速判断配置没配错。
验证顺利的话，把这套配置固化到集群的默认 Spark 配置里，后面所有作业自动生效。当时我是在部署脚本里统一加了一段配置块，方便后续新节点上线自动带上。
4. 性能测试与调优实战
4.1 测试方法论：别拿一条 SQL 说事
很多团队验证性能时喜欢找一条慢 SQL 跑一下，看提升多少。但这样得出的结论没有统计意义，因为大数据场景下作业的执行时间受数据倾斜、并发、缓存命中率影响非常大。我建议用一组覆盖不同场景的 SQL 做对比，比如分组聚合、大表 Join、排序取 Top N、带过滤的扫描。每个场景跑三遍取中位数，避免偶发因素干扰。
数据集我用的是一套模拟业务数据的表，规模在几十 GB 左右，分区数几十个。对比基线是原 Spark 默认配置跑出来的结果，对照组是开启 Comet 但保持其他参数不变。这个时候你就能比较客观地看到不同算子的收益差异了。
4.2 实测数据与观察
从我这边的实测看，三类典型场景提升非常明显。第一类是纯扫描加过滤，数据量大的时候提升通常在 1.8 到 2.2 倍。这个容易理解，native 扫描可以一次性处理一个 batch，SIMD 加速效果好，还省了 Java 对象创建的开销。第二类是带聚合的查询，比如按维度分组求汇总值，提升也明显，因为 HashAggregate 在 native 引擎里可以做到非常紧凑的哈希表，内存占用小，缓存效果好。第三类是 Shuffle 密集的 Join 场景，Comet 的 Shuffle 写入读取如果启用 native 实现，可以减少序列化开销，整体提升大概在 1.3 到 1.6 倍。
也有不那么理想的情况。数据量比较小的查询，比如只有几百 MB，收益非常有限，甚至因为 JNI 初始化和批处理建立有固定开销，还可能出现轻微回退。所以如果你的作业大部分是小查询，引入 Comet 的意义不大，更多是增加维护成本。
4.3 调优参数组合推荐
调优方向重点是内存和 Shuffle。我最终固定下来的配置大概是这样：开启spark.comet.enabled和spark.comet.exec.enabled，执行模式选 auto，开启 offheap 内存模式，堆外内存给 executor 内存的 30% 左右，Shuffle 的 native 实现开启。在批量测试中这套配置能稳定带来收益。
另一个可以调节的参数是 vectorized 批次的大小，一般是以行的数量来衡量，比如 2048、4096、8192。批次太小会放大 JNI 调用的开销，批次太大则缓存命中率下降。我自己测试下来 8192 比较合适，但不同数据宽窄结构效果不一样，建议用一个全表扫描 SQL 做一次梯度实验。
还有一个值得关注的参数与动态分配相关。因为 native 内存不受 Spark 的 Unified Memory 管理，开启 Executor 动态分配时，不能完全按原内存比例申请。这里需要观察 Spark UI 里 Storage Memory 和 Executor 的实际用量，如果频繁出现 Executor 因为内存不足被杀掉，就要适度调低spark.comet.memory.overhead.factor之类的参数，给 JVM 留出足够空间。
5. 常见问题与排查技巧实录
5.1 排障总思路
Comet 排障最大的难点在于它横跨 JVM 和 native 两个世界。日志输出的位置分散，有 JVM 侧的异常堆栈，也有 native 侧的 log 输出。我通常按这个顺序排查：先看 Spark UI 里有没有任务失败和异常类型；再看 executor 的 stdout/stderr 日志，重点搜索 "Comet" 关键字；如果涉及 JNI 加载失败，再到系统层面用ldd检查动态库依赖。
定位到具体算子后，通过 EXPLAIN 看执行计划里有没有 Comet 节点。如果某个预期会被 native 化的算子没有变成 Comet 节点，大概率是这个算子或表达式不在支持范围内，或者配置开关没开到位。这个时候不要硬调，先确认版本的算子支持列表，或者换个等价写法。
5.2 高频问题速查
我遇到最多的问题集中在几个方面。第一个是类加载类异常，比如ClassNotFoundException: comet.spark.*，原因通常是 jar 包没有放到正确位置，或者--jars只对 driver 生效，executor 上没有拿到。解决办法是把 jar 放到所有节点同名目录，或用 Spark 的 archive 机制统一分发。
第二个是 native 库加载失败，典型报错是Could not load library: libcomet.so。检查一下库文件权限、路径配置和 glibc 版本。我遇到过一个问题，动态库编译机器的 glibc 版本高于集群环境，导致运行时直接报错，解决办法是在编译服务器上调整兼容性，或直接使用官方提供的预编译包。
第三个是内存溢出。Comet 的 native 内存超限时，可能表现为 Executor 被系统杀掉，而不是普通 Java OOM。需要在日志里找是否存在 native 侧的内存日志，确认是不是 offheap 配置过小。我调这类问题一般就是调大 offheap，同时检查并行度，因为并行 Task 太多时，每个 Task 分配到的内存太小，批次就没法做大，反而会触发频繁 JNI 调用。
5.3 我的避坑清单
这里把几个容易反复踩的坑单独拎出来说。第一，不要随意升级 Comet 的小版本而不重新测试，因为执行计划的划分逻辑和算子支持列表随版本变化很快，可能上一个版本不用回退的查询，新版本里突然回退了，性能反而变差。第二个建议是，配置里尽量用显式的模式而不是默认值，至少把spark.comet.exec.mode固定下来，否则升级后默认值变动可能导致行为不一致。
第三个经验是关于和现有调优参数的联动。如果你之前的习惯是把spark.sql.adaptive.coalescePartitions.enabled打开，开启 Comet 后要重新验证一遍。因为 AQE 动态调整分区数和 native 执行计划的划分可能互相影响，配额缩并的逻辑和原来不一样。一次版本升级后有个作业运行时间反而翻倍，排查结果就是 AQE 的分区合并和 Comet 的 Shuffle 读取逻辑产生了不理想的交互，关掉 AQE 后问题消失。不过这是特定版本下的情况，建议以你自己环境的实测为准。
还有一点，Java 版本尽量和官方推荐保持一致。有些发行版自带高版本 JDK，但 Comet 的部分代码在 JDK 17 和 JDK 11 上表现不完全相同，尤其是反射访问模块的行为。我用的 JDK 8 没有遇到过问题，但如果你在 JDK 17 上碰到奇怪的 JNI 访问异常，先考虑降级 JDK 或对照官方支持矩阵。
6. 对落地的一些个人体会
DataFusion Comet 并不是万能的提速神器，它最适合的场景是 SQL 复杂、数据量大、执行时间长的离线批处理作业。如果你的作业已经用上了很多 Spark 特有优化技巧，比如手动调分区、精心设计的 Bucket 表，Comet 带来的额外收益可能会被稀释。但反过来，如果作业一直放在默认配置下没有精细调过，Comet 往往能带来一个不小的初始性能跳跃，因为它的 native 执行器在很多基础操作上比代码生成的 JVM 实现跑得更有效率。
从运营成本角度看，引入 Comet 主要增加的是部署适配和排障成本。它的架构决定了它不是一个纯黑盒，出了问题需要同时懂 Spark 任务管理和 native 层面的基础概念。对于团队里只要有一两个能看懂执行计划和 JNI 日志的人，这个成本完全可控。
我个人的建议是分三步走：先在测试环境用生产数据集的采样跑一遍对比；确认收益后再挑两三个核心作业做灰度；最后再逐步扩大到全量作业。整个过程里，保持和社区版本同步更新很重要，因为每次新版本大概率会修复一些回退场景和内存相关的 bug。这套方案设计得比较务实，它没有要求你推翻 Spark，而是像一把针对特定场景的螺丝刀，用好了能解决实际问题，用错场景也只是多一个工具而已。
