为什么在重新划分Spark数据帧时会有这么多空分区？

Question

我想将数据帧“df1”划分为3列。此数据帧正好有990个针对这3列的唯一组合：

In [17]: df1.createOrReplaceTempView("df1_view")

In [18]: spark.sql("select count(*) from (select distinct(col1,col2,col3) from df1_view) as t").show()
+--------+                                                                      
|count(1)|
+--------+
|     990|
+--------+

为了优化这个数据帧的处理，我想对df1进行分区，以获得990个分区，每个分区对应一个密钥：

In [19]: df1.rdd.getNumPartitions()
Out[19]: 24

In [20]: df2 = df1.repartition(990, "col1", "col2", "col3")

In [21]: df2.rdd.getNumPartitions()
Out[21]: 990

我写了一个简单的方法来计算每个分区中的行数：

In [22]: def f(iterator):
    ...:     a = 0
    ...:     for partition in iterator:
    ...:         a = a + 1
    ...:     print(a)
    ...: 

In [23]: df2.foreachPartition(f)

我注意到，实际上我得到的是628个带有一个或多个键值的分区，以及362个空分区。

我假设spark会以一种均匀的方式（1个键值=1个分区）重新分区，但这似乎不是这样，我觉得这种重新分区增加了数据倾斜，尽管它应该是另一种方式。。。

Spark用于在列上划分数据帧的算法是什么？有没有办法实现我认为可能实现的目标？

我在Cloudera上使用Spark 2.2.0。

Answer 1

要跨分区分发数据，Spark需要以某种方式将列的值转换为分区的索引。Spark中有两个缺省分区-Hash分区器和Range分区器。Spark中的不同转换可以应用不同的分区-例如连接将应用哈希分区。

基本上，对于散列分区器来说，将值转换为分区索引的公式应该是value。hashCode（）%numOfPartitions。在您的情况下，多个值映射到同一个分区索引。

如果想要更好的分发，可以实现自己的分区器。更多关于它的信息就在这里，这里，这里。