Spark Gluten 向量化执行框架

背景

向量化执行模型已在现代数据库引擎被广泛应用，例如ClickHouse、TiDB、StarRocks等。为进一步提升计算性能，充分利用CPU计算能力，大数据计算引擎的不同厂商也针对向量化计算进行多样化探索。

Databricks 2022年推出商业化产品Photon，旨在增强Spark计算而写的Native C++ 向量化计算引擎。其执行层架构如图所示9，Photon向量化计算基于最底层的Task调度粒度处理 。扩展新的转换规则将原物理计划转为Photon物理计划，在Task级别实现向量化计算，提供高效的湖仓一体计算模式。Photon向量化执行引擎取代了已有的Spark执行模式（JVM & Codegen）。

Gluten项目由Intel和Kyligence于2021年合作共建，旨在通过开源社区推进Native Spark实现，为Spark扩展向量化执行(Vectorized Execution)能力，进一步提升引擎查询性能和CPU利用率。Gluten整体设计思想与 Photon相似，沿用Spark执行框架，基于新增GlutenPlugin 和 加载Session Extensions实现自定义扩展。

整体架构

Gluten Plugin在向量化执行中处于“承上启下”位置，承上对接Spark物理计划，启下对接原生Native执行引擎。 Gluten Plugin组件如图所示包括[1]:

1. Plan Conversion: Gluten最核心组件，基于规则将SparkPlan转换为GlutenPlan；
2. Memory Manager: 实现Native Engine内存统一由Spark管理；
3. Columnar Shuffle: Gluten重要特性，通过列式数据结构优化 Spark Shuffle实现，避免了在 Shuffle 过程中频繁的数据行转列和列转行操作，从而显著提升Shuffle I/O性能；
4. Shim Layer: 支持多版本Spark;
5. Fallback: 执行容错和支持回退JVM执行；
6. Metric: 指标上报 Spark Metrics System。

Spark Driver处理查询请求的基本流程如下图所示，(1).基于Spark Catalyst框架进行SQL解析和逻辑计划优化，(2).基于SparkPlanner进行物理计划优化，(3).优化后的物理计划SparkPlan基于规则进行处理并转换成RDD对象。Spark底层执行基于火山模型实现，需要对每个算子进行迭代计算。

为提升执行效率，从 Spark 2.0 开始，社区开始引入全阶段代码生成(Whole-stage Code Generation)优化。基于CodeGen 将子计划树压缩为一段可执行代码，避免子计划树节点的逐个计算，大幅减少虚函数调用，显著提升查询性能。从 Spark 3.0 开始，社区支持自适应查询执行(Adaptive Query Execution, AQE) ，在DAG Stage执行过程中，基于上一个Stage的真实执行统计信息，重新生成更优的执行计划，动态优化下一个Stage的执行逻辑。

Gluten实现向量化计算的主要变更如下所示，在QueryExecution执行前优化过程中，通过注入规则对物理计划进行扩展处理，转为Gluten物理计划，使用向量化执行模式替换已有的JVM + CodeGen的执行模式。在Task执行阶段，基于Gluten RDD，通过JNI调用Native引擎(默认Velox)实现向量化执行。

实现细节

扩展流程

基于Gluten branch-1.1 分支，整体解析实现细节如下，主要包括四个模块: Spark、Gluten、Substrait和Velox，三个扩展流程: 规则注入、转换、执行。

1. 规则注入: 基于Spark Session Extensions能力，在初始化Gluten Plugin时加载Gluten Sessions的扩展注入器，核心注入器ColumnarOverrides提供列式执行转换规则，Gluten列式转换规则扩展自Spark定义的列式规则基类ColumnarRule。
2. 转换: 基于Catalyst和SparkPlanner优化得到的物理计划无法直接提交给Executor执行，需转换为RDD对象。在QueryExecution 中，由prepareForExecution方法进行转换RDD前的正确性处理和执行优化。 其中preparations维护处理规则，核心转换规则包括: (1). EnsureRequirements: 核心逻辑在ensureDistributionAndOrdering方法中，确保物理计划的数据分布(Distribution)与排序(Ordering)正确性，当数据分布和有序性无法满足当前节点的处理逻辑，则会添加Shuffle或排序操作；即添加Exchange节点或SortExec节点，其中，Exchange节点是实现数据并行化的重要算子。 (2). CollapseCodegenStages: 代码生成规则，将支持代码生成的计划树节点整合在一起，并添加WholeStageCodegenExec节点标识，执行时生成一段整合后的可执行代码，减少虚函数调用。针对不支持的代码生成的节点，会添加InputAdapter节点进行标识处理。 (3). ApplyColumnarRulesAndInsertTransitions: 基于ColumnarRule扩展的列式转换规则，转换成支持并加速列式数据读写的SparkPlan。转换操作分为三部分: preColumnarTransitions、insertTransitions、postColumnarTransitions。基于pre前置转换处理，根据输出的数据格式处理行转列(RowToColumnarExec)或列转行(ColumnarToRowExec)的数据插入转换。post后置转换处理可实现计划清理或全局执行效率优化。Gluten扩展ColumnarOverrideRules规则集，基于preOverrides 将SparkPlan转为GlutenPan(Transformer)，后置处理中，校验是否支持向量化计算，若不支持则回退处理，若支持则进一步将行转列&列转行相关的GlutenPlan转换为NativeConversion，关联特定Native引擎的物理计划。为提升执行效率，Gluten基于ColumnarCollapseTransformStages 进行全阶段转换优化，其参考CollapseCodegenStages 设计思想，将支持Native转换的操作整合在一起，并添加WholeStageTransformer节点，不支持Native执行的节点，则添加InputAdapter节点标识处理。
3. 执行: 在Spark执行前，通过toRdd方法将SparkPlan物理计划转为RDD，GlutenPlan重载doExecuteColumnar方法支持列式计算，Gluten WholeStageTransformer转换为Gluten RDD时，会将Spark计划子树转换为Substrait PB计划树，维护并生成对应的 Gluten Columnar RDD对象。在Executor Task 执行阶段，基于JNI调用原生引擎Velox执行。

作业执行

Spark从触发执行到真正作业执行主要分为两个阶段:

1. Driver端作业调度: 将Job作业转换为由RDD构成的DAG图，DAGScheduler基于RDD Shuffle操作(宽依赖)将DAG划分为不同的Stage (ShuffleMapStage & ResultStage)，一个Stage可由一组计算逻辑相同但完全独立的Task任务组成，TaskScheduler基于submitTasks触发Task级别资源调用。
2. Executor端任务执行: 一个Executor进程支持Task并行执行，每个Task基于runTask方法执行并处理各自的Partition数据。Gluten RDD在compute计算中，会基于JNI调用底层Native引擎实现向量化计算，并严格控制 JNI 调用次数，整个数据流转基于 Spark 原生 ColumnarBatch 抽象。

Shuffle是影响执行性能的重要一环，对于ShuffleMapTask任务，Spark提供统一的基于行数据模型的ShuffleManager管理。由于向量化计算采用列式数据结构，需要在 Shuffle Write 阶段引入数据列转行处理，在 Shuffle Read 阶段引入数据行转列处理。为提升列式数据的Shuffle读写性能，Gluten扩展ColumnarShuffleManager实现基于列式数据模型Shuffle管理。

Substrait是基于Protocol Buffersr Plan定义跨语言关系代数的序列化协议，核心是提供SQL计划树的规范，使得这些计划树可以在不同的执行引擎之间转换和优化，解决不同引擎间互操作性差的问题。Gluten选型Substrait 作为跨运行环境/跨语言的计划转换协议，生成Gluten RDD时，将SparkPlan物理计划转为Substrait PB计划树；Task执行时，再将Substrait计划树转为Velox计划树执行。

附录

1. 向量化执行引擎框架 Gluten 宣布正式开源，并亮相 Spark 技术峰会