Spark HashPartitioner 意外分区答案

【问题标题】：Spark HashPartitioner Unexpected PartitioningSpark HashPartitioner 意外分区
【发布时间】：2017-01-26 05:28:35
【问题描述】：

我正在使用HashPartioner，但得到了意想不到的结果。我使用 3 个不同的字符串作为键，并将分区参数设置为 3，所以我希望有 3 个分区。

val cars = Array("Honda", "Toyota", "Kia")

val carnamePrice = sc.parallelize(for {
 x <- cars
 y <- Array(100,200,300)
} yield (x, y), 8)
val rddEachCar = carnamePrice.partitionBy(new HashPartitioner(3))
val mapped =   rddEachCar.mapPartitionsWithIndex{
                (index, iterator) => {
                   println("Called in Partition -> " + index)
                   val myList = iterator.toList

                   myList.map(x => x + " -> " + index).iterator
                }
             }
mapped.take(10)

结果如下。它只提供 2 个分区。我检查了字符串的哈希码（69909220 75427 -1783892706）。这里可能有什么问题？可能我误解了分区算法。

Array[String] = Array((Toyota,100) -> 0, (Toyota,200) -> 0, (Toyota,300) -> 0, (Honda,100) -> 1, (Honda,200) -> 1, (Honda,300) -> 1, (Kia,100) -> 1, (Kia,200) -> 1, (Kia,300) -> 1)

【问题讨论】：

标签： apache-spark rdd

【解决方案1】：

这里没有什么奇怪的。 HashPartitioner使用的Utils.nonNegativeMod实现如下：

def nonNegativeMod(x: Int, mod: Int): Int = {
  val rawMod = x % mod
  rawMod + (if (rawMod < 0) mod else 0)
}

使用 3 个分区，键分布定义如下：

for { car <- Seq("Honda", "Toyota", "Kia") } 
  yield (car -> nonNegativeMod(car.hashCode, 3))

Seq[(String, Int)] = List((Honda,1), (Toyota,0), (Kia,1))

这正是你在你的情况下得到的。换句话说，没有直接哈希冲突并不能保证没有以任意数字为模的冲突。

【讨论】：