引言：本文將全面且深入地解析 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 與設置不同。