Spark Parquet向量化读取原理与性能调优实战

1. 从一次查询变慢说起：为什么需要向量化读取

去年帮一个团队排查一个Spark作业的性能问题，场景很典型：一张按天分区的Parquet宽表，单分区大概几十GB，字段数在两百左右，查询只取其中五六个列做聚合。按理说列式存储加列裁剪，扫描量应该很小，但实际跑起来每个task的CPU时间高得离谱，火焰图里一大半时间耗在了org.apache.parquet.column.impl.ColumnReaderImpl相关的方法上，而不是在真正的聚合计算里。

这个问题最后指向的就是Parquet读取路径。Spark早期读Parquet走的是"逐行"模式：一行一行地把记录拼出来，再交给上层算子。列存格式本来是按列连续存放的，逐行读等于把列存的优势又还回去了——每读一行都要在多个列reader之间来回跳，虚函数调用、边界检查、对象创建全挤在热路径上。向量化读取要解决的就是这件事：不再一行一行地处理，而是一批一批（通常1024行）地按列读取、按列计算，把JIT能优化的紧循环暴露出来，把每行的固定开销摊薄到一批上。

这篇内容我打算把Spark里Parquet向量化读取这条链路拆开讲清楚：它到底"向量化"了什么、底层依赖哪些组件、列批是怎么被组装出来的、哪些条件会触发退化回逐行模式，以及在实际调优时我会关注哪些参数和现象。适合已经会用Spark、但想搞清楚"为什么我的Parquet查询快/慢"的读者，也适合正在做数据湖查询引擎选型的人。文中涉及的具体类名和调用链以Spark 3.x的常见实现为参考，不同小版本会有差异，但主干逻辑是一致的。

2. 向量化到底"向量"了什么：列批与列式内存布局

2.1 逐行读取的固定开销从哪里来

要理解向量化的收益，先得看清逐行模式的成本结构。在ColumnarBatch出现之前，Spark读Parquet的核心是SpecificParquetRecordReaderBase配合生成的RecordMaterializer，每调用一次nextKeyValue()，内部大致要做这些事：

• 对每个被投影的列，调用对应的ColumnReader，从页（Page）的解码缓冲区里读出一个值；
• 处理definition level和repetition level，判断这个值是不是null、是不是嵌套结构里的重复项；
• 把读出的原始类型装箱成Java对象（比如int变成Integer，或者写进一个可复用的InternalRow）；
• 组装成一行，交给上层。

单看一次调用没多少活，但乘以几亿行就是灾难。更麻烦的是这些操作里有大量分支判断和虚函数调用，JIT很难把它们内联成一个紧凑的循环，CPU的分支预测和指令缓存命中率都不好看。我见过的一个极端例子是：同样一份数据，只把读取路径从逐行换成向量化，CPU时间直接降了60%以上，聚合逻辑一行没改。

2.2 列批：把"一批行"当成一个处理单元

向量化的核心数据结构是列批（ColumnarBatch），你可以把它理解成"一个二维的小表"：行数是固定的（默认spark.sql.parquet.columnarReaderBatchSize，默认值1024），列数等于本次扫描实际需要的列数。每一列是一个ColumnVector，底层通常是一块连续的内存（on-heap的int[]、long[]，或者off-heap的MemoryBlock）。

关键在于，ColumnVector暴露的接口是按列、按批的，比如getInts(int rowId, int count, int[] target)这种批量读取方法。上层算子（比如HashAggregateExec的向量化版本）拿到列批后，也是按列做循环：对一个int列求和，就是一个for (int i = 0; i < batch.numRows(); i++) sum += intCol[i];这样的紧循环。这种循环对JIT极其友好，能自动向量化（SIMD）、能消除边界检查、能常驻寄存器。

用生活化的类比：逐行读取像是去超市买东西，每拿一件商品就跑去收银台结一次账；向量化则是推着购物车一次性拿一批，最后统一结账。结账这个动作本身没变，但来回跑的次数少了几个数量级。

2.3 为什么是1024这个数量级

批大小不是随便定的。太小，摊薄不了固定开销；太大，CPU的L1/L2缓存放不下，反而会频繁cache miss。1024行乘以每行几十字节，大概几十KB到上百KB，正好落在L2缓存的舒适区。这个值可以通过spark.sql.parquet.columnarReaderBatchSize调整，但我实测下来，除非你的单行特别宽或者特别窄，否则默认值基本就是最优区间，调它收益很小，不建议作为调优首选。

3. 读取链路上的三个关键角色

3.1 ParquetFileFormat：决定走不走向量化路径

Spark读Parquet的入口是ParquetFileFormat。它在buildReaderWithPartitionValues里会根据一系列条件判断是否启用向量化读取，核心判断逻辑大致是：

• 表本身是否支持（比如某些老格式、某些嵌套类型不支持）；
• 查询里有没有BinaryType、DecimalType等需要特殊处理的类型；
• 是否开启了相关开关（spark.sql.parquet.enableVectorizedReader，默认true）；
• 分区列的处理方式。

只有全部满足，才会构造VectorizedParquetRecordReader；否则回退到ParquetRecordReader。这个判断很重要，因为很多人以为"我开了向量化就一定走向量化"，实际上一个不支持的字段类型就能让整条路径退化。我踩过的坑：一张表里有个decimal(38,18)的列，查询里虽然没用到它，但因为schema里存在，早期版本会直接让整个reader退化，后来才改成按投影列判断。

3.2 VectorizedParquetRecordReader：批的调度者

VectorizedParquetRecordReader是向量化路径的"大脑"。它的nextBatch()方法负责把一批数据填进ColumnarBatch。内部维护了一个ParquetColumnVector数组，每个元素对应一个被投影的列，负责该列的读取和填充。

它的工作流程可以概括为：

1. 检查当前批是否还有剩余行，没有就触发一次nextBatch；
2. 对每个ParquetColumnVector，调用其readBatch方法，把数据从Parquet页解码到ColumnVector里；
3. 处理缺失列（比如分区列、常量列），用ConstantColumnVector填充；
4. 返回组装好的ColumnarBatch。

这里有个容易被忽略的细节：批的边界和Parquet的row group边界不是对齐的。一个row group可能有几万行，而批是1024行，所以reader需要维护跨批的状态，处理"上一批读了一半的页"这种情况。这也是为什么向量化reader的状态机比逐行reader复杂得多。

3.3 ParquetColumnVector：单列的读取与解码

ParquetColumnVector是真正干活的人。它持有一个VectorizedColumnReader（来自Parquet-mr库），负责把某一列的数据从页里解码出来。它的readBatch方法大致做这几件事：

• 如果当前页读完了，调用readNextPage加载下一页；
• 根据页的编码方式（PLAIN、RLE_DICTIONARY、DELTA_BINARY_PACKED等），调用对应的解码器；
• 处理definition level，把null的位置在ColumnVector里标记出来（通过setNull）；
• 把解码后的值批量写入ColumnVector的底层数组。

这里最影响性能的是字典编码。Parquet默认对低基数列用字典编码，读取时如果字典能装进内存，解码就是"查表"——从字典里按索引取值，非常快。但如果字典太大（比如高基数的字符串列），reader会退化成逐值解码，性能会明显下降。我在实际项目里遇到过：一个user_id列基数上千万，字典页几百MB，读取时内存和CPU都吃紧，后来改成对这类列关闭字典编码（写入时设置parquet.enable.dictionary=false），读取反而更稳。

4. 从Parquet页到ColumnVector：解码过程拆解

4.1 页的类型与解码路径

Parquet文件里，一列的数据被切成若干页，页分三种：数据页（DataPage V1/V2）、字典页（DictionaryPage）、索引页。读取时，reader先读字典页（如果有），再逐个读数据页。

数据页V1和V2的区别在于：V1把repetition level、definition level和值混在一起存；V2把它们分开存，并且对level用了更紧凑的编码。V2在读取时通常更快，因为level的解码可以独立出来，但兼容性上V1更普遍。Spark的向量化reader对两者都支持，走的是不同的解码分支。

解码一个数据页的大致步骤：

1. 读取页头，拿到编码方式、值的个数、level的编码方式；
2. 如果是字典编码，先读字典页建立索引到值的映射；
3. 解码repetition level和definition level（如果是嵌套列）；
4. 根据definition level判断哪些位置是null，哪些有值；
5. 解码实际的值，写入ColumnVector。

4.2 definition level与null的处理

这是向量化读取里比较绕的一块。Parquet用definition level表示"这个值在嵌套结构里到底存不存在"。对于非嵌套的顶层列，definition level只有0和1：0表示null，1表示有值。对于嵌套列，level会更复杂。

向量化reader处理null的方式是：先根据definition level算出一个"null bitmap"，然后只对非null的位置解码值。这样ColumnVector里null位置的值是未定义的（通常是0），但通过isNull标记位能正确识别。上层算子在做计算时必须先检查null，这也是为什么向量化的聚合算子代码里到处是if (!col.isNullAt(i))。

注意：如果你自己写代码直接访问ColumnVector的底层数组，一定要先判断null，否则会把null位置当成0参与计算，结果静默出错。这个坑我在做自定义UDF时踩过。

4.3 字典解码为什么快，什么时候会变慢

字典编码的本质是"用整数索引代替原始值"。假设一列有100万个字符串值，但只有1000个不同的值，那么存储时只存1000个字符串的字典，加上100万个int索引。读取时，解码就是value = dictionary[index]，一次数组访问。

快的原因有两个：一是索引是定长的int，解码循环非常紧凑；二是字典本身可以复用，不用为每个值创建新的String对象。

变慢的场景也很明确：

• 字典太大：字典本身要全部加载进内存，如果字典有几百MB，光是加载和缓存就会拖慢；
• 基数太高：如果几乎每个值都不同，字典索引的压缩收益趋近于零，解码时还要多做一次查表，反而不如直接读原始值；
• 字典页跨row group：每个row group有自己的字典，切换row group时要重新加载。

我的经验是：对基数低于10万的字符串列，字典编码几乎总是划算的；超过百万级别，就要考虑关闭字典，或者用更合适的类型（比如把字符串ID转成long）。

5. 哪些情况会让向量化"退化"

5.1 类型不支持：最常见的退化原因

不是所有Parquet类型都能向量化。以Spark 3.x为例，以下情况通常会导致退化：

我遇到最多的是嵌套类型导致的退化。比如一个array<struct<...>>的列，如果嵌套层级深，reader可能无法用向量化方式展开，只能回退。这时候的排查方法是看Spark UI的SQL页签，或者直接看物理计划里FileScan parquet节点有没有ColumnarBatch相关的标记。

5.2 分区列与常量列的处理

分区列不在Parquet文件里，是目录名解析出来的。向量化reader对分区列的处理是构造一个ConstantColumnVector，整批都是同一个值。这本身没问题，但如果分区列参与了复杂的谓词下推，可能会影响批的构造效率。一般来说分区列数量不多（几十个以内）影响可以忽略。

5.3 批大小与内存的权衡

前面提过批大小默认1024。如果单行特别宽（比如几百个列，每列都是长字符串），一个批可能占几十MB内存，多个批并发就会给GC造成压力。这时候可以适当调小批大小，比如512甚至256。反过来，如果单行很窄（就几个int列），可以调大到2048或4096，进一步摊薄开销。但如前所述，这个参数的调优空间不大，优先解决类型退化和字典问题。

6. 实测对比：向量化到底能快多少

6.1 测试环境与数据集构造

为了给出一个可参考的数字，我构造了一个模拟数据集：一张宽表，200个列，其中180个是int/long，20个是string（基数约5万），按天分区，单分区约5000万行。用Parquet写入，默认字典编码。查询是"取5个int列做sum和count，按一个低基数列分组"。

对比两组配置：

• A组：spark.sql.parquet.enableVectorizedReader=true（默认）；
• B组：spark.sql.parquet.enableVectorizedReader=false。

其他参数保持一致，executor数量、内存、并行度都相同。

6.2 结果与解读

扫描字节数相同说明IO不是瓶颈，差异全在CPU和对象创建上。B组GC时间高，是因为逐行模式为每行创建了大量临时对象。A组的CPU时间优势比总耗时优势更大，说明向量化主要省的是计算，当作业是IO瓶颈时，收益会小很多。

这个结果也提醒一件事：向量化不是万能的。如果你的查询本身就是IO密集（比如全表扫描大字段），或者瓶颈在网络shuffle，开不开向量化差别不大。它的主战场是"扫描量不大但行数多、计算密集"的场景。

6.3 一个反直觉的发现

在上面这个测试里，我额外试了一组：把string列的字典编码关掉。结果A组的耗时从42s涨到了51s。原因是字典编码在读取时确实更快，关掉后虽然写入的文件略小，但读取解码变慢了。这印证了前面的结论：字典编码对读取性能通常是正向的，不要轻易关闭，除非字典本身大到内存扛不住。

7. 调优时我会盯住的几个点

7.1 先确认有没有真的走向量化

这是第一步，也是最容易被跳过的一步。方法有几个：

• 看Spark UI的SQL页签，展开FileScan parquet节点，看Output里有没有ColumnarBatch；
• 在物理计划里找VectorizedParquetRecordReader相关的类名；
• 最直接的是看task的CPU时间分布，如果大量时间在ColumnReaderImpl，基本就是逐行。

我习惯用第三种，因为火焰图不会骗人。

7.2 字典与编码的取舍

写入端的编码选择会直接影响读取端。我的经验规则：

• 低基数（<10万）字符串列：保持字典编码；
• 高基数（>100万）字符串列：考虑关闭字典，或者转成数值类型；
• 数值列：默认编码通常够用，除非有特殊的delta编码需求。

这些可以在写入时通过Parquet的writer参数控制，也可以在表级别设置。

7.3 批大小与并行度的配合

批大小影响单task的内存占用，并行度影响同时运行的task数。两者要一起看：如果批大小是1024、单批占10MB，并行度是200，那么峰值内存就是2GB左右（还没算其他开销）。如果内存紧张，优先调小批大小，而不是降并行度，因为降并行度会直接拉长作业时间。

7.4 嵌套列的处理策略

如果表里有复杂的嵌套列，而查询又用不到，最有效的办法是在写入时就把它拍平或者拆出去。向量化reader对嵌套的支持一直在改进，但再好的支持也比不上"根本不需要读它"。我在一个项目里把一张宽表的嵌套字段拆成独立的表，读取性能提升了近一倍，因为主表变窄了，列裁剪更彻底。

8. 写在最后的一点个人体会

Parquet向量化读取这件事，表面看是"换个reader"，实际上它牵动的是整个查询链路的配合：写入端的编码选择、schema的设计、查询的投影和谓词、批大小和内存的平衡。我见过太多人只盯着"开没开向量化"这一个开关，却忽略了上游的编码和schema才是决定性的。

如果只能给一条建议，那就是：先把schema和编码理顺，再谈向量化。一张设计良好的窄表，配上合适的字典编码，即使在某些边缘场景退化成逐行，性能也不会太难看；反过来，一张几百列、嵌套深、编码混乱的表，就算向量化全开，也救不回来。向量化是放大器，不是万能药。