Blaze RangePartitioning 算子Native实现全解析
引言：本文将全面且深入地解析Blaze RangePartitioning算子的Native实现过程。相较于原生Spark，RangePartitioning的Native实现在执行时间上达到了30%的显著下降，同时在资源开销方面节省了高达76%。这一改进大幅降低了运行成本，展现了Native实现带来的巨大优势。
一、算子描述
RangePartitioning是shuffle partitioning的一种分区类型。它通过根据数据的值范围将数据划分成多个分区。每个分区包含特定范围内的值，通常用于处理有序的数据集，能够根据数据值进行动态划分。
RangePartitioning 的基本思想是：先对数据采样找到划分标志bounds，根据bounds将数据划分成多个近似大小的区间，然后将数据按所属区间写入对应partition，用于order by 全排序场景。
二、实现方案
RangePartitioning 实现主要包含采样和partition划分两个部分。
步骤一：首先需要获取每个partition对应的区间划分范围bounds，所以会先对全量数据进行采样，算出 partitionNum - 1 个区间分割点bounds。具体流程如下：
1、在driver端基于InternalRow进行数据采样：
-
通过spark.sql.execution.rangeExchange.sampleSizePerPartition参数控制每个分区平均采样数量，设置一个稍微过采样一点的采样数sampleSizePerPartition。
-
对每个分区采用蓄水池采样（Reservoir Sampling）算法进行采样。
-
对采样结果评估，记录采样不均衡的分区重新采样（某个分区数据量过多，按照sampleSizePerPartition均值采样会出现样本数少于实际应采样数量，即采样不均衡的情况）。
-
计算每个样本的权重weight，通过sumWeights/numReducer = step找到每个边界的步长，类似于直方图划分边界找出numReducer-1个分割点bounds。
2、由于采样数据量可能不足导致bounds较少，需要重新设置partitionNum=bounds.len + 1。因此会出现RangePartitioning的实际partition num与设置数量不同的情况。
3、定义rangepartition的序列化方式，主要包括三个参数：SortExpr、numPartitions、Bounds。进而转成native 算子进行后续处理。
步骤二：在native端需要再计算一次全量数据，将数据按分割点bounds写入对应的partition。具体流程如下：
1、将bounds和input数据都转成可直接比较的arrow-row类型。
2、针对每个batch，对将数据与bounds进行比较并确定所在partition id：
-
如果bounds.len<=128，直接进行比较。
-
如果bounds.len>128，进行二分查找提速。
三、优化效果
通过构造sql语句测试加速效果：
sql测试例子
11.8GB数据量：
insert overwrite table blaze_t.like_lineitem select * from tpch_parquet_1000.lineitem order by l_quantity
实现Native RangePartitioning
执行计划：
sql 时间1073.516 s
Stage Total Time Across All Tasks: 8.9h
没有实现Native RangePartitioning，会回退到spark的RangePartitioning
sql 时间 1357.814 s
Stage Total Time Across All Tasks 38.1h
| Blaze | Spark | |
| sql总耗时 | 1073.516 s | 1357.814 s |
| Stage Total Time Across All Tasks | 8.9h | 38.1h |
| Stage时间提升 | (38.1-8.9)/38.1 = 0.7664042 | |
多个不同sql测试取均值
Stage时间提升：76.94%
四、总结
-
多次测试取均值，RangePartitioning 实现native相比旧版执行时间下降30%，资源开销节约70%
-
由于采样结果可能较少导致bounds小于partition num-1，RangePartitioning可能实际执行的partition num与设置不同。