Spark向量化计算在美团生产环境的实践15

 1 什么是向量化计算

1.1 并行数据处置惩罚：SIMD指令

让我们从一个简单问题开始：假设要实现“数组a+b存入c”，设三个整型数组的长度都是100，那么只需将“c = a + b”置于一个100次的循环内，代码如下：

1. void addArrays(const int* a, const int* b, int* c, int num) {
2.   for (int i = 0; i < num; ++i) {
3.     c[i] = a[i] + b[i];
4.   }
5. }

复制代码

我们知道：计算在CPU内完成，逻辑计算单元利用寄存器中的数据，算术运算的源利用数要先放置到CPU的寄存器中，哪怕简单的内存拷贝也需要过CPU寄存器。以是，完成“c = a + b”需经三步：

• 加载（Load），从内存加载2个源利用数（a和b）到2个寄存器。
  
• 计算（Compute），实行加法指令，作用于2个寄存器里的源利用数副本，结果产生到目标寄存器。
  
• 存储（Store），将目标寄存器的数据存入（拷贝）到目标内存位置（c）。
  

其中，加载和存储对应访存指令（Memory Instruction），计算是算术加指令，循环实行100次上述三步调，就完成了“数组a + 数组b => 数组c”。该流程即对应传统的计算架构：单指令单数据（SISD）顺序架构，恣意时间点只有一条指令作用于一条数据流。如果有更宽的寄存器（超机器字长，比如256位16字节），一次性从源内存同时加载更多的数据到寄存器，一条指令作用于寄存器x和y，在x和y的每个分量（比如32位4字节）上并行举行加，并将结果存入寄存器z的各对应分量，最后一次性将寄存器z里的内容存入目标内存，那么就能实现单指令并行处置惩罚数据的结果，这就是单指令多数据（SIMD）。

图1：向量化计算“数组a+b存入c”
单指令多数据对应一类并行架构（当代CPU一般都支持SIMD实行），单条指令同时作用于多条数据流，可成倍的提升单核计算能力。SIMD非常得当计算密集型使命，它能加速的根本缘故原由是“从一次一个凌驾到一次一组，从而实现用更少的指令数完成同样的计算使命。”
1996年，Intel推出的X86 MMX（MultiMedia eXtension）指令集扩展可视为SIMD的起点，随后演进出SSE（1999年）SSE2/3/4/5、AVX（2008）/AVX2（2013）、AVX512（2017）等扩展指令集。在linux体系中可以通过lscpu或cpuid命令查询CPU对向量化指令的支持环境。
1.2 向量化实行框架：数据局部性与运行时开销

实行引擎常规按行处置惩罚的方式，存在以下三个问题：

• CPU Cache命中率差。一行的多列（字段）数据的内存紧挨在一起，哪怕只对其中的一个字段做利用，其他字段所占的内存也需要加载进来，这会抢占稀缺的Cache资源。Cache命失会导致被请求的数据从内存加载进Cache，等候内存利用完成会导致CPU实行指令停息（Memory Stall），这会增长延时，还大概浪费内存带宽。
  
• 变长字段影响计算效率。假设一行包括int、string、int三列，其中int类型是固定长度，而string是变长的（一般表示为int len + bytes content），变长列的存在会导致无法通过行号算offset做快速定位。
  
• 虚函数调用带来额外开销。对一行的多列举行处置惩罚通常会封装在一个循环里，会抽象出一个类似handle的接口（C++虚函数）用于处置惩罚某类型数据，各字段类型会override该handle接口。虚函数的调用多一步查表，且无法被内联，循环内高频调用虚函数的性能影响不可忽视。
  

图2：row by row VS blcok by block
因此，要让向量化计算发挥威力，只使用SIMD指令还不敷，还需要对实行框架层面举行改造，变Row By Row为Block By Block：

• 数据按列组织替换按行组织（在Clickhouse和Doris里叫Block，Velox里叫Vector），这将进步数据局部性。参与计算的列的多行数据会内存紧凑的生存在一起，CPU可以通过预取指令将接下来要处置惩罚的数据加载进Cache，从而减少Memory Stall。不参与计算的列的数据不会与被处置惩罚的列竞争Cache，这种内存交互的隔离能进步Cache亲和性。
  
• 同一列数据在循环里被施加雷同的计算，批量迭代将减少函数调用次数，通过模版能减少虚函数调用，低落运行时开销。针对固定长度类型的列很轻易被并行处置惩罚（通过行号offset到数据），这样的实行框架也有利于让编译器做主动向量化代码生成，显着减少分支，减轻推测失败的惩罚。结合模板，编译器会为每个实参生成特定实例化代码，避免运行时查找虚函数表，并且由于编译器知道了具体的类型信息，可以对模板函数举行内联睁开。
  

图3：向量化实行框架示例
1.3 怎样使用向量化计算

• 主动向量化（Auto-Vectorization）。当循环内没有复杂的条件分支，没有数据依赖，只调用简单内联函数时，通过编译选项（如gcc -ftree-vectorize、-O3），编译器可以将顺序实行代码翻译成向量化实行代码。可以通过观察编译hint输出和反汇编确定是否生产了向量化代码。
  
  
  
  • 编译hint输出，编译：g++ test.cpp -g -O3 -march=native -fopt-info-vec-optimized，实行后有类似输出“test.cpp:35:21: note: loop vectorized”。
    
  • 反汇编，gdb test + (gdb) disassemble /m function_name，看到一些v打头的指令（比方vmovups、vpaddd、vmovups等）。
    
  
  
• 使用封装好的函数库，如Intel Intrinsic function、xsimd等。这些软件包中的内置函数实现都使用了SIMD指令举行优化，相当于high level地使用了向量化指令的汇编，详见：https://www.intel.com/content/www/us/en/docs/intrinsics-guide/index.html。
  
• 通过asm内嵌向量化汇编，但汇编指令跟CPU架构相干，可移植性差。
  
• 编译器暗示：
  
  
  
  • 使用编译指示符（Compiler Directive），如Cilk（MIT开辟的用于并行编程的中央层编程语言和库，它扩展了C语言）里的#pragma simd和OpenMP里的#pragma omp simd。
    
  • Compiler Hint。通过__restrict去修饰指针参数，告诉编译器多个指针指向不雷同不重叠的内存，让编译器放心大胆的去优化。
    
  
  
• 如果循环内有复杂的逻辑或条件分支，那么将难以向量化处置惩罚。
  

以下是一个向量化版本数组相加的例子，使用Intel Intrinsic function：

1. #include <immintrin.h> // 包含Intrinsic avx版本函数的头文件
3. void addArraysAVX(const int* a, const int* b, int* c, int num) {
4.   assert(num % 8 == 0); // 循环遍历数组，步长为8，因为每个__m256i可以存储8个32位整数
5.   for (int i = 0; i < num; i += 8) {  
6.     __m256i v_a = _mm256_load_si256((__m256i*)&a[i]); // 加载数组a的下一个8个整数到向量寄存器
7.     __m256i v_b = _mm256_load_si256((__m256i*)&b[i]); // 加载数组b的下一个8个整数到向量寄存器
8.     __m256i v_c = _mm256_add_epi32(v_a, v_b); // 将两个向量相加，结果存放在向量寄存器
9.     _mm256_store_si256((__m256i*)&c[i], v_c); // 将结果向量存储到数组c的内存
10.   }
11. }
13. int main(int argc, char* argv[]) {
14.   const int ARRAY_SIZE = 64 * 1024;
15.   int a[ARRAY_SIZE] __attribute__((aligned(32))); // 按32字节对齐，满足某些向量化指令的内存对齐要求
16.   int b[ARRAY_SIZE] __attribute__((aligned(32)));
17.   int c[ARRAY_SIZE] __attribute__((aligned(32)));
18.   srand(time(0));
19.   for (int i = 0; i < ARRAY_SIZE; ++i) {
20.     a[i] = rand(); b[i] = rand(); c[i] = 0; // 对数组a和b赋随机数初始值
21.   }
23.   auto start = std::chrono::high_resolution_clock::now();
24.   addArraysAVX(a, b, c, ARRAY_SIZE);
25.   auto end = std::chrono::high_resolution_clock::now();
26.   std::cout << "addArraysAVX took " << std::chrono::duration_cast<std::chrono::microseconds>(end - start).count() << " microseconds." << std::endl;
27.   return 0;
28. }

复制代码

addArraysAVX函数中的_mm256_load_si256、_mm256_add_epi32、_mm256_store_si256都是Intrinsic库函数，内置函数命名方式是

• 利用浮点数：_mm(xxx)_name_PT
  
• 利用整型：_mm(xxx)_name_epUY
  

其中(xxx)代表数据的位数，xxx为SIMD寄存器的位数，若为128位则省略，AVX提供的__m256为256位；name为函数的名字，表示功能；浮点内置函数的后缀是PT，其中P代表的是对矢量（Packed Data Vector）还是对标量（scalar）举行利用，T代表浮点数的类型（若为s则为单精度浮点型，若为d则为双精度浮点）；整型内置函数的后缀是epUY，U表示整数的类型（若为无符号类型则为u，否在为i），而Y为利用的数据类型的位数。
编译：g++ test.cpp -O0 -std=c++11 -mavx2 -o test。选项-O0用于禁用优化（因为开启优化后有大概主动向量化），-mavx2用于启用AVX2指令集。
测试发现：非向量化版本addArrays耗时170微秒，而使用Intrinsic函数的向量化版本addArraysAVX耗时58微秒，耗时降为原来的1/3。
2 为什么要做Spark向量化计算

从业界发展环境来看，近几年OLAP引擎发展迅速，该场景追求极致的查询速率，向量化技术在Clickhouse、Doris等Native引擎中得到广泛使用，降本增效的趋势也渐渐扩展到数仓生产。2022年6月DataBricks发表论文《Photon- A Fast Query Engine for Lakehouse Systems》，Photon是DataBricks Runtime中C++实现的向量化实行引擎，相比DBR性能均匀提升4倍，并已应用在Databricks商业版上，但没有开源。2021年Meta开源Velox，一个C++实现的向量化实行库。2022 Databricks Data & AI Summit 上，Intel 与Kyligence先容了合作开源项目Gluten，旨在为Spark SQL提供Native Vectorized Execution。Gluten+Velox的组合，使Java栈的Spark也可以像Doris、Clickhouse等Native引擎一样发挥向量化实行的性能上风。
从美团内部来看，数仓生产有数万规模计算节点，很多业务决策依赖数据及时产出，若应用向量化实行技术，在不升级硬件的环境下，既可得到可观的资源节省，也能加速作业实行，让业务更快看到数据和做出决策。根据Photon和Gluten的公开数据，应用向量化Spark实行效率至少可以提升一倍，我们在物理机上基于TPC-H测试Gluten+Velox相Spark 3.0也有1.7倍性能提升。

图4：Gluten+Velox在TPC-H上的加速比，来自Gluten
3 Spark向量化计算怎样在美团实施落地

3.1 团体建设思路

• 更关注资源节省而不但追求实行加速。Spark在美团主力场景是离线数仓生产，与OLAP场景不同，时间相对不敏感，但资源（内存为主）基数大成本敏感。离线计算历史已久，为充实利用存量服务器，我们不能依赖硬件加速的本事如更多的内存、SSD、高性能网卡。我们评估收益的核心指标是总「memory*second」低落。
  
• 基于C++/Rust等Native语言而非Java举行开辟。Java语言也在向量化实行方面做实行，但JVM语言对底层控制力弱（如无法直接内嵌SIMD汇编），再加上GC等固有缺陷，还远远谈不上成熟，而体系向的语言（C/C++、Rust）则成为挖掘CPU向量化实行潜能的首选。
  
• 可插拔、面向多引擎而非绑定Spark。固然面向不同工作负载的各类大数据引擎层出不穷，但其架构分层则相似，一般包括编程接口、实行操持、调理、物理实行、容错等，尤其实行算子部分有较多可复用实现。如Meta内部重要大数据引擎有Presto和Spark，建设一个跨引擎的实行库，优化同时支持Presto和Spark显然是更好的选择；OLAP引擎向量化计算本身就是标配；流计算引擎出于性能思量，也可以攒批而非一条条处置惩罚数据（mini batch），因此向量化实行也有性能提升空间。我们认为面向不同场景操持的大数据引擎，有大概共用同一个高性能向量化实行库。
  
• 使用开源方案而非完全自研。Spark有几百个function和operator，向量化改造的工作量巨大，从性能、完成度、适配成本、是否支持多引擎、社区的活跃度等方面综合思量，我们终极选择了Gluten+Velox的方案。
  
• 迁移过程对用户透明，保证数据一致。Spark的几百个function和 operator都要通过C++重新实现，同时还涉及Spark、Gluten、Velox版本变革，很轻易实现出现偏差导致计算结果不一致的环境。我们开辟了一个用于升级验证的黑盒测试工具（ETL Blackbox Test），可以将一个作业运行在不同版本的实行引擎上举行端到端验证，包括实行时间、内存及CPU资源使用环境、作业数据的对比结果（通过对比两次实行的行数，以及每一列所有数据md5的加和值来确定数据是否一致）。
  

3.2 Spark+Gluten+Velox计算流程

通过Spark的plugin功能，Gluten将Spark和向量化实行引擎（Native backend，如Velox）连接起来，分为Driver plugin和Executor Plugin。在Driver端，SparkContext初始化时，Gluten的一系列规则（如ColumnarOverrideRules）通过Spark Extensions注入，这些规则会对Spark的实行操持举行校验，并把Gluten支持的算子转换成向量化算子（如FileScan会转换成NativeFileScan），不能转换的算子上报Fallback的缘故原由，并在回退的部分插入Column2Row、Row2Column算子，生成Substrait实行操持。在Executor端，接收到Driver侧的LaunchTask RPC消息传输的Substrait实行操持后，再转换成Native backend的实行操持，终极通过JNI调用Native backend实行。
Gluten希望能尽大概多的复用原有的Spark逻辑，只是把计算部分转到性能更高的向量化算子上，如作业提交、SQL解析、实行操持的生成及优化、资源申请、使命调理等行为都还由Spark控制。

图5：Spark+Gluten+Velox架构图
3.3 阶段分别

在我们开始Spark向量化项目时，开源版本的Gluten和Velox还没有在业界Spark生产环境大规模实践过，为了低落风险最小代价验证可行性，我们把落地过程分为以下五个阶段逐步举行：

• 软硬件适配环境确认。Velox要求CPU支持bmi、bmi2、f16c、avx、avx2、sse指令集，需要先确定服务器是否支持；在生产环境运行TPC-DS或者TPC-H测试，验证理论收益；公司内部版本适配，编译运行，跑通典型使命。当时Gluten只支持Spark3.2和Spark3.3，思量到Spark版本升级成本更高，我们暂时将相干patch反打到Spark3.0上。这个阶段我们办理了大量编译失败问题，发起用社区保举的OS，在容器中编译&运行；如果要在物理机上运行，需要把相干依赖部署到各个节点，或者使用静态链接的方式（开启vcpkg）。
  

1. cat /proc/cpuinfo | grep --color -wE "bmi|bmi2|f16c|avx|avx2|sse"

复制代码

• 稳固性验证。确定测试集，完善稳固运行需要的feature，以到达比较高的测试通过率，包括支持ORC、Remote shuffle、HDFS适配、堆内堆外的内存配置等。本阶段将测试通过率从不足30%提升到90%左右。
  
• 性能收益验证。由于向量化版本和原生Spark分别使用堆外内存和堆内内存，引入翻倍内存的配置，以及一些高性能feature支持不完善，一开始生产环境测试性能结果不及预期。我们逐个分析办理问题，包括参数对齐、去掉arrow中央数据转换、shuffle batch fetch、Native HDFS客户端优化、使用jemelloc、算子优化、内存配置优化、HBO适配等。本阶段将均匀资源节省从-70%提升到40%以上。
  
• 一致性验证。重要是问题修复，对所有非SLA作业举行大规模测试，筛选出稳固运行、数据完全一致、有正收益的作业。
  
• 灰度上线。将向量化实行环境发布到所有服务器，对符合条件的作业分批上线，增长监控报表，网络收益，对性能不及预期、发生数据不一致的作业及时回退原生Spark上。此过程用户无感知。
  

整个实施过程中，我们通过收益转化漏斗找到收益最大的优化点，指导项目迭代。下图为2023年某一时期的相邻转化环境。

图6：Spark向量化项目收益转化漏斗图
4 美团Spark向量化计算遇到的挑战

4.1 稳固性问题

• 聚合时Shuffle阶段OOM。在Spark中，Aggregation一般包括Partial Aggregation、Shuffle、Final Aggregation三个阶段，Partial Aggregation在Mapper端预聚合以低落Shuffle数据量，加速聚合过程、避免数据倾斜。Aggregation需要维护中央状态，如果Partial Aggregation占用的内存凌驾肯定阈值，就会提前触发Flush同时后续输入数据跳过此阶段，直接进入ShuffleWrite流程。Gluten使用Velox默认配置的Flush内存阈值（Spark堆外内存*75%），由于Velox里Spill功能还不敷完善（Partial Aggregation不支持Spill），这样大作业场景，后续ShuffleWrite流程大概没有足够的内存可以使用(可用内存<25%*Spark堆外内存)，会引起作业OOM。我们接纳的策略是通过在Gluten侧调低Velox Partial Aggregation的Flush阈值，来低落Partial Aggregation阶段的内存占用，避免大作业OOM。这个方案在可以让大作业运行通过，但是理论上提前触发Partial Aggergation Flush会低落Partial Aggretation的结果。更合理的方案是Partial Aggregation支持Spill功能，Gluten和Velox社区也一直在完善对向量化算子Spill功能的支持。
  
• SIMD指令crash。Velox对数据复制做了优化，如果该类型对象是128bit（比如LongDecimal类型），会通过SIMD指令用于数据复制以提升性能。如下图所示，Velox库的FlatVector<T>::copyValuesAndNulls()函数里的一行赋值会调用T:perator=()，调用的movaps指令必须作用于16B对齐的地址，不满意对齐要求会crash。我们在测试中复现了crash，通过日志确定有未按16B对齐的地址出现。无论是Velox内存池还是Gluten内存池分配内存都强制做了16B对齐，终极确认是Arrow内存池分配出的地址没对齐（Gluten用了三方库Arrow）。这个问题可以通过为LongDecimal类型重载operator=利用符修复，但这样做大概影响性能，也不彻底，因为不能清除还有其他128bit类型对象存在。终极我们与Gluten社区修改了Arrow内存分配策略，强制16B对齐。
  

图7：Crash代码示例
4.2 支持ORC并优化读写性能

Velox的DWIO模块原生只支持DWRF和Parquet两种数据格式，美团大部分表都是基于ORC格式举行存储的。DWRF文件格式是Meta内部所接纳的ORC分支版本，其文件结构与ORC相似，比如针对ORC文件的不同区域，可通过复用DWRF的Reader来完成相干数据内容的读取。

图8：Dwrf文件格式

• DwrfReader：用于读取文件层面的元数据信息，包括PostScript、Footer和Header。
  
• DwrfRowReader：用来读取StripeFooter，以便确定每个column的Stream位置。
  
• FormatData：用来读取StripeIndex，从而确定每个RowGroup的位置区间。
  
• ColumnReader：用来读取StripeData，完成不同column的数据抽取。
  

我们完善了Velox ORC格式的支持，并对读取链路做了优化，重要工作包括：

• 支持RLEv2解码（Velox-5443）并在解码过程中完成Filter下推（Velox-6647）。我们将Apache RLEv2解码逻辑移植到了Velox，通过BMI2指令集来加速varint解码过程中的位运算，并在解码过程中下推过滤不须要的数据。
  
• 支持Decimal数据类型（Velox-5837）以及该数据类型的Filter下推处置惩罚（Velox-6240）。
  
• ORC文件句柄复用以低落HDFS的NN处置惩罚压力（Velox-6140）。出于线程安全层面的思量，HdfsReadFile每次pread都会开启一个新文件句柄来做seek+read，客户端会向NameNode发送大量open请求，加重HDFS的压力。我们通过将文件的读取句柄在内部做复用处置惩罚（thread_local模式），减少向NN发送的open请求。
  
• 使用ISA-L加速ORC文件解压缩。我们对ORC文件读取耗时trace分析得出，zlib解压缩占总耗时60%，解码占30%，IO和其他仅占10%，解压效率对ORC文件读取性能很关键。为此，我们对ZlibDecompressor做了重构，引入Intel的解压缩向量化库ISA-L来加速解压缩过程。
  

基于这些优化，改造后的Velox版ORC Reader读取时间减少一半，性能提升一倍。

图9：Apache ORC与改造后的Velox ORC读取性能对比，上为Apache ORC
4.3 Native HDFS客户端优化

首先先容一下HDFS C++客户端对ORC文件读取某一列数据的过程。第一步，读取文件的最后一个字节来确定PostScript长度，读取PostScript内容；第二步，通过PostScript确定Footer的存储位置，读取Footer内容；第三步，通过Footer确定每个Stripe的元数据信息，读取StripeFooter；第四步，通过StripeFooter确定每个Column的Stream位置，读取需要的Stream数据。

图10：ORC文件读取过程
在生产环境测试中，我们定位到两个数据读取相干的性能问题：

• 小数据量随机读放大。客户端想要读取[offset ~ readEnd]区间内的数据，发送给DN的实际读取区间却是[offset ~ endOfCurBlock]，导致[readEnd ~ endOfCurBlock]这部分数据做了无效读取。这样操持重要是为了优化顺序读场景，通过预读来加快后续访问，然而针对随机读场景（小数据量下比较广泛），该方式却适得其反，因为预读出的数据很难被后续使用，增长了读放大行为。我们优化为客户端只向DN传递需要读取的数据区间，DN侧不提前预取，只返回客户端需要的数据。
  

图11：读放大过程示意图

• DN慢节点导致作业运行时间变长。我们发现很多大作业的HDFS长尾耗时非常高，HDFS的均匀read时延只有10ms左右，P99.99时延却到达了6秒，耗时最长的请求甚至到达了5分钟，但在不启用EC场景下，HDFS的每个block会有三副本，完全可以切换到空闲DN访问。为此我们对客户端的读请求链路做了重新的操持与调解，实时监测每个DN的负载环境，基于P99.9分位请求时延判定慢节点，并将读请求路由到负载较低的DN上面。
  

HDFS Native客户端读优化之后，均匀读写耽误低落了2/3，吞吐提升2倍。
4.4 Shuffle重构

Gluten在shuffle策略的支持上，没有预留好接口，每新增一种shuffle模式需要较大改动。美团有自研的Shuffle Service，其他公司也大概有本身的Shuffle Service（如Celeborn），为了更好适配多种shuffle模式，我们提议对shuffle接口重新梳理，并主导了此讨论和操持。
Gluten中的shuffle抽象第一层是数据格式（Velox是RowVector，Gluten引入的Arrow是RecordBatch），第二层是分区方式（RoundRobin、SinglePart、Hash、Range），如果要支持新shuffle模式（local、remote），需要实现2*4*2=16个writer，将会有大量冗余代码。分区具体实现应该与数据格式和shuffle模式无关，我们用组合模式替换继承模式。另外，我们在shuffle中直接支持了RowVector，避免Velox和Arrow对应数据类型之间的额外转换开销。

图12：重构前后shuffle模块UML对比
4.5 适配HBO

HBO（Historical Based Optimization）是通过作业历史运行过程中资源的实际使用量，来推测作业下一次运行需要的资源并设置资源相干参数的一种优化本事。美团已往在原生Spark上通过调配堆内内存取得了8%左右的内存资源节省。
Gluten重要使用堆外内存（off-heap），这与原生Spark重要使用堆内内存（on-heap）不同。初期出于稳固性思量Gluten和原生Spark的运行参数团体一致，总内存大小雷同，Gluten off-heap 占比75%， on-heap占比25%。这样配置既会导致内存利用率不高（原生Spark的内存使用率58%，向量化版作业内存使用率 38%），也会使一部分作业on-heap内存配置偏低，在算子回退时导致使命OOM。
我们把HBO策略推广到堆外内存，向量化计算的内存节省比例从30%提升到40%，由于heap内存配置不合理的OOM问题全部消除。

图13：HBO流程图
4.6 一致性问题

• 低版本ORC数据丢失。hive-0.13之前使用的ORC，Footer信息不包含列名，只有ID用来表示第几列（如Col1, Col2…）。Velox TableScan算子在扫表的时候，如果下推的Filter里包含IsNotNull(A)，会根据列名A查找该列数据，由于无法匹配到列名，会误判空文件，导致数据缺失。Spark在生成读ORC表的实行操持时，通过访问HiveMetaStore得到表的Schema信息，并在物理算子FileSourceScanExec中生存了表的Schema信息。Gluten对该算子举行doTransform()转换时，会把表的Schema信息序列化到Substrait的ReadRel里。接下来的Substrait操持转Velox操持阶段，我们把表的Schema信息传给Velox的HiveTableHandle，在构造Velox的DwrfReader时修正ORC文件Footer里的Schema信息（如果Footer的Schema不包含列名，就读取表Schema里的对应列的名称举行赋值），办理了这个问题。
  
• count distinct结果错误。比如这样一条SQL：select A, B, count(distinct userId), sum(amt) from t group by 1,2 ，Gluten会把count(distinct userId) 变为count(userId)，通过把userId加到GroupingKey里来实现distinct语义。具体处置惩罚过程如下：
  

表1：示例SQL在Spark中的处置惩罚步调
在第3步的Intermediate Aggregation中，为了节省内存和加速实行，当Velox的HashAggregate算子满意触发Flush的条件时（HashTable内存占用凌驾阈值或者聚合结果低于阈值），Velox会标记 partialFull=true，触发Flush利用（计算HashTable里已经缓存数据的Intermediate Result），并且后续输入的数据不再实行Intermediate Aggregation的计算，直接进入第4步的Partial Aggregation。如果后续输入的数据里包含重复的userId，count(userId)会因为去重不彻底而结果错误。我们短期的修复方案是禁用Intermediate Aggregation的提前Flush功能，直到所有数据都输入之后再获取该阶段的聚合结果。
这个方案的弊端有两个：1）HashTable的内存占用会变大，需要同时开启HashAggregate算子的Spill功能避免OOM；2）直接修改了Velox的HashAggregate算子内部代码，从Velox自身的角度来看，没有单独针对Distinct相干的聚合做处置惩罚，随着后续迭代，大概影响所有用到Intermediate Aggregation的聚合过程。
鉴于此，Gluten社区提供了一个更加均衡的办理方案，针对这类Distinct Aggregation，生成实行操持时，Spark的Partial Merge Aggregation不再生成Intermediate Aggregation，改为生成Final Aggregation（不会提前flush、不使用merge_sum），同时共同聚合函数的Partial Companion函数来返回Intermediate结果。这样就从实行操持转换策略层面规避这个问题，避免对Velox里Final Aggregation内部代码做过多的改动。

• 浮点类型转换精度错误。形如查询SELECT concat(col2, cast(max(col1) as string)) FROM (VALUES (CAST(5.08 AS FLOAT), 'abc_')) AS tab(col1, col2) group by col2; 在Spark中返回abc_5.08，在Gluten中返回abc_5.079999923706055。浮点数5.08不能用二进制分数准确表达，近似表示成5.0799999237060546875。Velox通过函数folly::to<std::string>(T val)来实现float类型到string类型的转换，这个函数底层依赖开源库google::double-conversion, folly里默认设置了输出宽度参数DoubleToStringConverter::SHORTEST(可以准确表示double类型的最小宽度)，转换时经过四舍五入之后返回 5.079999923706055。我们把宽度参数改为DoubleToStringConverter::SHORTEST_SINGLE(可以准确表示float类型的最小宽度)，转换时经过四舍五入之后返回 5.08。
  

5 上线结果

我们已上线了2万多ETL作业，均匀内存资源节省40%+，均匀实行时间减少13%，证明Gluten+Velox的向量化计算方案生产可行。向量化计算除了能进步计算效率，也能进步读写数据的效率，如某个作业的Input数据有30TB，已往需要实行7小时，绝大部份时间花在了读数据和解压缩上面。使用向量化引擎后，因为上文提到的ISA-L解压缩优化，列转行的开销节省，以及HDFS Native客户端优化，实行时间减少到2小时内。

图14：上线优化结果
6 未来规划

我们已上线向量化计算的Spark使命只是小部分，操持2024年能让绝大部分的SQL使命运行在向量化引擎上。
6.1 Spark向量化之后对开源社区的跟进策略

Spark、Gluten、Velox三个社区各有本身思量和版本发布节奏，从一个社区到多个社区的引擎维护复杂度上升。我们的应对有二，一是计算引擎有不同层次，Spark升级重要思量功能语义实现、实行操持、资源和task调理，Gluten和Velox的升级重要思量物理算子性能优化，各取所长；二是只管减少和社区的差别，公司内部适配只在Spark中实现，公司内的UDF以git submodule形式单独维护。

• 升级到Spark3.5。Gluten最低支持的Spark版本为3.2，23年我们为了低落验证成本，选择在Spark3.0兼容Gluten，但继承升级迭代成本比较高，在推广之前，应该升级到更新的Spark版本。Spark3.5将会是Gluten社区对Spark3.x上长期支持的稳固版本。高版本Spark也有一些额外收益，我们基于TPC-H实测，Spark3.5相比Spark3.0，「memory*second」减少40%，实行时间减少17%，根据之前升级经验，生产环境约莫能到达一半结果。
  
• 保持Spark版本长期稳固。高版本Spark对Hadoop版本的升级迭代带来比较高适配成本，内部迭代的feature也有比较高的迁移成本，因此我们均匀3年才会升级一次Spark版本，更多是将需要的feature归并到内部分支。
  
• 快速跟进Gluten/Velox新版本。升级到Spark3.5之后，我们内部Spark版本与Gluten社区的兼容性成本很低，并且向量化相干feature还会连续迭代，预计每半年可升级一次线上版本。
  

6.2 提升向量化覆盖率的策略

• 扩大向量化算子和UDF范围。我们整理了影响权重最高的几十个算子回退问题与Gluten社区一起办理，对于大量内部UDF，则会探索用大模型来将UDF批量改写为C++版本的向量化实现。
  
• 扩大File format支持向量化范围。美团内部有约20%的表为textfile格式，还有接近10%的表使用内部开辟的format，只能按行读取也不支持下推，加上行转列都会有额外性能开销，影响终极结果。我们将会把textfile全部转为ORC，为自研format提供C++客户端，进一步提升向量化计算性能。
  

from ：Spark向量化计算在美团生产环境的实践 - 美团技术团队
一样平常开辟Guava提效工具库核心实用指南梳理_guava string转list-CSDN博客

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。