scala - HashPartitioner 是如何工作的？

Question

我阅读了 HashPartitioner 的文档。不幸的是，除了 API 调用之外，没有任何解释。我假设 HashPartitioner 根据键的哈希对分布式集进行分区。例如，如果我的数据类似于

(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))
}

由于我们没有分区器，我们的数据集在分区之间均匀分布( Default Partitioning Scheme in Spark ):

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))

由于传递给 HashPartitioner 的参数定义了我们期望一个分区的分区数量:

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

由于rdd按关键数据分区，将不再均匀分布:

countByPartition(rddTwoP).collect()

Array[Int] = Array(2, 4)

因为有三个键并且只有两个不同的 hashCode 值 mod numPartitions 这里没有什么意外的:

(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))

最后，使用 HashPartitioner(7) 我们得到七个分区，其中三个非空分区，每个分区有 2 个元素:

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)

摘要和注释

• HashPartitioner 采用单个参数来定义分区数量

• 使用键的哈希将值分配给分区。 hash 函数可能因语言而异(Scala RDD 可能使用 hashCode，DataSets 使用 MurmurHash 3、PySpark、portable_hash)。

  在像这样的简单情况下，其中 key 是一个小整数，您可以假设 hash 是一个身份 (i = hash(i))。

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

• 如果 key 分布不均匀，您可能会遇到部分集群空闲的情况

• 键必须是可散列的。你可以查看我的回答A list as a key for PySpark's reduceByKey阅读有关 PySpark 特定问题的信息。 HashPartitioner documentation 突出显示了另一个可能的问题。 :

  Java arrays have hashCodes that are based on the arrays' identities rather than their contents, so attempting to partition an RDD[Array[]] or RDD[(Array[], _)] using a HashPartitioner will produce an unexpected or incorrect result.

• 在 Python 3 中，您必须确保散列是一致的。请参阅What does Exception: Randomness of hash of string should be disabled via PYTHONHASHSEED mean in pyspark?

• 哈希分区器既不是单射的也不是满射的。可以将多个键分配给单个分区，并且某些分区可以保留为空。

• 请注意，当前基于哈希的方法在与 REPL 定义的案例类 ( Case class equality in Apache Spark ) 结合使用时在 Scala 中不起作用。

• HashPartitioner(或任何其他Partitioner)对数据进行打乱。除非在多个操作之间重用分区，否则它不会减少要打乱的数据量。