哈希算法在 Dataset.repartition 中的工作原理

Question

我正在对数据集进行基本的重新分区。我在文件 test.csv

中有如下数据

abc,1
def,2
ghi,3
jkl,4
mno,5

我正在阅读像

这样的数据框

val df= spark.read.csv("test.csv")
val repart=df.repartition(5,col("_c1"))
repart.write.csv("/home/partfiles/")

现在写入数据后，它创建了 5 个部分文件，这是正确的。但在这个过程中，只有三部分文件具有如下所示的正确数据。

part00000 -empty
part00001 -jkl,4
part00002 -empty
part00003 -ghi,3
part00004 - abc,1
            def,2
            mno,5

但是因为我已经根据第 2 列重新分区并且所有数据都不同，理想情况下它应该创建 5 个不同的部分文件。
根据数据集 API 文档

Returns 一个由给定的分区表达式分区成 numPartitions 的新数据集。生成的数据集是散列分区的。

然后我用谷歌搜索了一些东西，发现了这篇关于分区的精彩文章 ()。
这篇文章提到DataSet使用Murmur3Hash算法。所以我写了小代码来根据这篇SO文章（How can I use Scala's MurmurHash implementation: scala.util.MurmurHash3?）获取哈希值。

class Murmur3{
  import scala.util.hashing.{ MurmurHash3 => MH3 }
  val values= (1 to 5).map(p=> p.toString)
  val result = values.map(n => (n,MH3.stringHash(n,MH3.stringSeed)))
  def resultVal(): Unit ={
    val dn= result.map( d=> d._1 -> (d._2,d._2 % 5)) //
    dn.foreach(println)
  }
}

这给了我这个价值。输出类似于 (number,(hasvalue, hashvalue%5))

(1,(-1672130795,0))
(2,(382493853,3))
(3,(1416458177,2))
(4,(1968144336,1))
(5,(2100358791,1))

现在必须根据此数据生成 4 个部分文件。但是如何生成 3 部分文件。请让我知道 hashpartitioning 在数据集的情况下是如何工作的。

Answer 1

您犯的错误是假设散列是在 Scala 字符串上完成的。在实践中，Spark 直接在不安全的字节数组上散列。

所以表达式等价于

import org.apache.spark.sql.functions.hash

Seq("1", "2", "3", "4", "5").toDF.select(
  when(hash($"value") % 5 > 0, hash($"value") % 5 )
    .otherwise(hash($"value") % 5 + 5)
).show
// +-----------------------------------------------------------------------------------------+
// |CASE WHEN ((hash(value) % 5) > 0) THEN (hash(value) % 5) ELSE ((hash(value) % 5) + 5) END|
// +-----------------------------------------------------------------------------------------+
// |                                                                                        4|
// |                                                                                        4|
// |                                                                                        3|
// |                                                                                        1|
// |                                                                                        4|
// +-----------------------------------------------------------------------------------------+

给出观察到的分布。

哈希算法在 Dataset.repartition 中的工作原理

How hashing algorithm works in Dataset.repartition

scala

apache-spark

apache-spark-dataset