HashPartitioner是如何工作的？

Question

(1,1), (1,2), (1,3), (2,1), (2,2), (2,3)

new HashPartitoner(numPartitions) //What does numPartitions do?

new HashPartitoner(1)
new HashPartitoner(2)
new HashPartitoner(10)

那么HashPartitioner实际上是如何工作的呢？

Answer 1

好吧，让我们让数据集稍微有趣一点：

val rdd = sc.parallelize(for {
    x <- 1 to 3
    y <- 1 to 2
} yield (x, None), 8)

我们有六个要素：

rdd.count

Long = 6

无分区程序：

rdd.partitioner

Option[org.apache.spark.Partitioner] = None

rdd.partitions.length

Int = 8

import org.apache.spark.rdd.RDD

def countByPartition(rdd: RDD[(Int, None.type)]) = {
    rdd.mapPartitions(iter => Iterator(iter.length))
}

countByPartition(rdd).collect()

Array[Int] = Array(0, 1, 1, 1, 0, 1, 1, 1)

import org.apache.spark.HashPartitioner
val rddOneP = rdd.partitionBy(new HashPartitioner(1))

rddOneP.partitions.length

Int = 1

countByPartition(rddOneP).collect

Array[Int] = Array(6)

请注意，洗牌后的值顺序是不确定的。

如果我们使用HashPartitioner(2)也是如此

val rddTwoP = rdd.partitionBy(new HashPartitioner(2))

我们将得到2个分区：

rddTwoP.partitions.length

Int = 2

countByPartition(rddTwoP).collect()

Array[Int] = Array(2, 4)

(1 to 3).map((k: Int) => (k, k.hashCode, k.hashCode % 2))

scala.collection.immutable.IndexedSeq[(Int, Int, Int)] = Vector((1,1,1), (2,2,0), (3,3,1))

rddTwoP.mapPartitions(iter => Iterator(iter.map(_._1).toSet)).collect()

Array[scala.collection.immutable.Set[Int]] = Array(Set(2), Set(1, 3))

val rddSevenP = rdd.partitionBy(new HashPartitioner(7))
rddSevenP.partitions.length

Int = 7

countByPartition(rddTenP).collect()

Array[Int] = Array(0, 2, 2, 2, 0, 0, 0)

在像这样的简单情况下，key是一个小整数，您可以假定hash是一个标识(i=hash(i))。

Scala API使用nonnegativeMod根据计算的哈希确定分区，

如果密钥分布不均匀，则可能会出现集群部分空闲的情况

哈希分区器既不是内射的，也不是满射的。可以将多个键分配给一个分区，有些分区可以保持为空。

请注意，当前基于哈希的方法在Scala中与REPL定义的case类（Apache Spark中的case类相等）结合时不起作用。

HashPartitioner（或任何其他Partitioner)洗牌数据。除非在多个操作之间重用分区，否则不会减少要洗牌的数据量。