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

见贤思齐

总第592篇 2024年第012篇Apache Spark是一个优秀的计算引擎，广泛应用于数据工程、机器学习等领域。向量化执行技术在不升级硬件的情况下，既可获得资源节省，又能加速作业执行。Gluten+Velox解决方案为Spark换上了向量化执行引擎，本文将阐述美团在这一方向的实践和思考。1 什么是向量化计算1.1 并行数据处理：SIMD指令1.2 向量化执行框架：数据局部性与运行时开销1.3 如何使用向量化计算2 为什么要做Spark向量化计算3 Spark向量化计算如何在美团实施落地3.1 整体建设思路3.2 Spark+Gluten+Velox计算流程3.3 阶段划分4 美团Spark向量化计算遇到的挑战4.1 稳定性问题4.2 支持ORC并优化读写性能4.3 Native HDFS客户端优化4.4 Shuffle重构4.5 适配HBO4.6 一致性问题5?上线效果6?未来规划6.1 Spark向量化之后对开源社区的跟进策略6.2 提升向量化覆盖率的策略7 致谢1 什么是向量化计算| 1.1 并行数据处理：SIMD指令让我们从一个简单问题开始：假设要实现“数组a+b存入c”，设三个整型数组的长度都是100，那么只需将“c[i] = a[i] + b[i]”置于一个100次的循环内，代码如下：void?addArrays(const?int*?a,?const?int*?b,?int*?c,?int?num)?{??for?(int?i?=?0;?i? 数组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：#include??//?包含Intrinsic?avx版本函数的头文件void?addArraysAVX(const?int*?a,?const?int*?b,?int*?c,?int?num)?{??assert(num?%?8?==?0);?//?循环遍历数组，步长为8，因为每个__m256i可以存储8个32位整数??for?(int?i?=?0;?i?(end?-?start).count()?<