本文将全面且深入地解析 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

多个不同 sql 测试取均值

Stage 时间提升：76.94%

四、总结

• 多次测试取均值，RangePartitioning 实现 native 相比旧版执行时间下降 30%，资源开销节约 70%

• 由于采样结果可能较少导致 bounds 小于 partition num-1，RangePartitioning 可能实际执行的 partition num 与设置不同。