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与设置不同。