Apache Spark 中的 RangePartitioner 是如何實現數據採樣的？

02-07

最近自己在學習 Apache Spark 的源碼，整體還挺順利，唯獨範圍分區器（Range Partitioner）這一塊一直卡殼，在此請教下各位知乎大神們。
問題主要集中在 RangePartitioner.rangeBounds 方法中。Range Partitioner 給我感覺很像是 TeraSort 演算法，先對父 RDD 分區中的數據進行採樣，樣本排序，確定範圍邊界，但是具體的實現又跟 TeraSort 有諸多不同。嗯，我的疑惑在於：
1. 採樣演算法方面，RangePartitioner 使用是水塘採樣法（Reservoir Sampling）。據我了解水塘採樣法是一種在線採樣演算法，主要應用於需要進行採樣的集合個數不可知的情況，可在 RangePartitioner 中集合個數 —— 也就是父 RDD 分區內數據記錄的個數是可知的，似乎沒必要採樣這種演算法，而且水塘採樣法還需要一個數據遍歷的過程，效率不是反倒變低了嗎？

2. RangePartitioner.rangeBounds 中，預計採樣的個數 sampleSize = math.min(20.0 * partitions, 1e6)，其中 partitions 是子 RDD 的分區個數，而分攤給父 RDD 中每個分區的採樣數 val sampleSizePerPartition = math.ceil(3.0 * sampleSize / rdd.partitions.size).toInt，其中乘以 3.0 的意義何在？
// This is the sample size we need to have roughly balanced output partitions, capped at 1M. val sampleSize = math.min(20.0 * partitions, 1e6)
// Assume the input partitions are roughly balanced and a little bit. val sampleSizePerPartition = math.ceil(3.0 * sampleSize / rdd.partitions.size).toInt
如上，感謝各位！

很細節的問題，從你的描述能看出你對這塊代碼和相關的背景知識也有一定的了解，但是對某些具體細節可能理解的稍有偏差。

數據採樣是低成本分析大數據特性的常用方法，數據採樣的方式有很多種，最常用的是隨機採樣，即保證每個數據元素被採樣的概率是相同的。隨機採樣也有不同的實現方式，如根據數據元素被採樣後是否放回總體，或者根據一定的概率進行採樣還是根據確定的樣本大小進行採樣。在Spark的RangePartitioner實現中採用的Reservoir Sampler演算法，是一種確定樣本大小，採樣後不放回總體的隨機採樣方法，這個採樣演算法可以在不知道總體大小的情況下，只需一次遍歷總體即可隨機採樣到確定大小的樣本。其實Reservoir Sampler演算法一般會比隨機概率採樣演算法效率低很多，Spark RangePartitioner的實現可能是考慮到採樣分區結果對接下來RDD的計算性能影響很大，所以在這兒採用效率低一點的採樣演算法，來保證確定的樣本大小。

對於你的問題：

1. 首先，在對RDD分區進行遍歷前是不知道RDD每個分區數據量大小的，RangePartitioner只對RDD進行了一次遍歷，就是採用Reservoir Sampler進行數據採樣的時候，代碼如下：

val sketched = rdd.mapPartitionsWithIndex { (idx, iter) =&> val seed = byteswap32(idx ^ (shift &<&< 16)) val (sample, n) = SamplingUtils.reservoirSampleAndCount( iter, sampleSizePerPartition, seed) Iterator((idx, n, sample)) }.collect()

2. 在這兒乘3的原因是RDD的分區可能有數據傾斜，sampleSize是期望的樣本大小，但是某些分區的數據量可能少於sampleSize/PartitionNumber，乘以3後期望其他的分區可以多採樣點數據，使得總的採樣量達到或超過sampleSize。