引言：本文將全面且深入地解析 Blaze RangePartitioning 算子的 Native 實(shí)現(xiàn)過(guò)程。相較于原生 Spark，RangePartitioning 的 Native 實(shí)現(xiàn)在執(zhí)行時(shí)間上達(dá)到了 30%的顯著下降，同時(shí)在資源開(kāi)銷方面節(jié)省了高達(dá) 76%。這一改進(jìn)大幅降低了運(yùn)行成本，展現(xiàn)了 Native 實(shí)現(xiàn)帶來(lái)的巨大優(yōu)勢(shì)。

一、算子描述

RangePartitioning 是 shuffle partitioning 的一種分區(qū)類型。它通過(guò)根據(jù)數(shù)據(jù)的值范圍將數(shù)據(jù)劃分成多個(gè)分區(qū)。每個(gè)分區(qū)包含特定范圍內(nèi)的值，通常用于處理有序的數(shù)據(jù)集，能夠根據(jù)數(shù)據(jù)值進(jìn)行動(dòng)態(tài)劃分。

RangePartitioning 的基本思想是：先對(duì)數(shù)據(jù)采樣找到劃分標(biāo)志 bounds，根據(jù) bounds 將數(shù)據(jù)劃分成多個(gè)近似大小的區(qū)間，然后將數(shù)據(jù)按所屬區(qū)間寫(xiě)入對(duì)應(yīng) partition，用于 order by 全排序場(chǎng)景。

二、實(shí)現(xiàn)方案

RangePartitioning 實(shí)現(xiàn)主要包含采樣和 partition 劃分兩個(gè)部分。

步驟一：首先需要獲取每個(gè) partition 對(duì)應(yīng)的區(qū)間劃分范圍 bounds，所以會(huì)先對(duì)全量數(shù)據(jù)進(jìn)行采樣，算出 partitionNum - 1 個(gè)區(qū)間分割點(diǎn) bounds。具體流程如下：

1、在 driver 端基于 InternalRow 進(jìn)行數(shù)據(jù)采樣：

• 通過(guò) spark.sql.execution.rangeExchange.sampleSizePerPartition 參數(shù)控制每個(gè)分區(qū)平均采樣數(shù)量，設(shè)置一個(gè)稍微過(guò)采樣一點(diǎn)的采樣數(shù) sampleSizePerPartition。
• 對(duì)每個(gè)分區(qū)采用蓄水池采樣（Reservoir Sampling）算法進(jìn)行采樣。
• 對(duì)采樣結(jié)果評(píng)估，記錄采樣不均衡的分區(qū)重新采樣（某個(gè)分區(qū)數(shù)據(jù)量過(guò)多，按照 sampleSizePerPartition 均值采樣會(huì)出現(xiàn)樣本數(shù)少于實(shí)際應(yīng)采樣數(shù)量，即采樣不均衡的情況）。
• 計(jì)算每個(gè)樣本的權(quán)重 weight，通過(guò) sumWeights/numReducer = step 找到每個(gè)邊界的步長(zhǎng)，類似于直方圖劃分邊界找出 numReducer-1 個(gè)分割點(diǎn) bounds。

2、由于采樣數(shù)據(jù)量可能不足導(dǎo)致 bounds 較少，需要重新設(shè)置 partitionNum=bounds.len + 1。因此會(huì)出現(xiàn) RangePartitioning 的實(shí)際 partition num 與設(shè)置數(shù)量不同的情況。

3、定義 rangepartition 的序列化方式，主要包括三個(gè)參數(shù)：SortExpr、numPartitions、Bounds。進(jìn)而轉(zhuǎn)成 native 算子進(jìn)行后續(xù)處理。

步驟二：在 native 端需要再計(jì)算一次全量數(shù)據(jù)，將數(shù)據(jù)按分割點(diǎn) bounds 寫(xiě)入對(duì)應(yīng)的 partition。具體流程如下：

1、將 bounds 和 input 數(shù)據(jù)都轉(zhuǎn)成可直接比較的 arrow-row 類型。

2、針對(duì)每個(gè) batch，對(duì)將數(shù)據(jù)與 bounds 進(jìn)行比較并確定所在 partition id：

• 如果 bounds.len<=128，直接進(jìn)行比較。
• 如果 bounds.len>128，進(jìn)行二分查找提速。

三、優(yōu)化效果

通過(guò)構(gòu)造 sql 語(yǔ)句測(cè)試加速效果：

sql 測(cè)試?yán)?/strong>

11.8GB 數(shù)據(jù)量：

insert overwrite table blaze_t.like_lineitem select * from tpch_parquet_1000.lineitem order by l_quantity

復(fù)制代碼

實(shí)現(xiàn) Native RangePartitioning

執(zhí)行計(jì)劃：

sql 時(shí)間 1073.516 s

Stage Total Time Across All Tasks: 8.9h

沒(méi)有實(shí)現(xiàn) Native RangePartitioning，會(huì)回退到 spark 的 RangePartitioning

sql 時(shí)間 1357.814 s

Stage Total Time Across All Tasks  38.1h

多個(gè)不同 sql 測(cè)試取均值

Stage 時(shí)間提升：76.94%

四、總結(jié)

• 多次測(cè)試取均值，RangePartitioning 實(shí)現(xiàn) native 相比舊版執(zhí)行時(shí)間下降 30%，資源開(kāi)銷節(jié)約 70%
• 由于采樣結(jié)果可能較少導(dǎo)致 bounds 小于 partition num-1，RangePartitioning 可能實(shí)際執(zhí)行的 partition num 與設(shè)置不同。