ToB企服应用市场:ToB评测及商务社交产业平台

标题: Spark向量化计算在美团生产环境的实践15 [打印本页]

---

作者: 美丽的神话    时间: 2025-1-11 04:33
标题: 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企服之家，中国第一个企服评测及商务社交产业平台。

---

欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/)

Powered by Discuz! X3.4