Spark：增加分区的数量而不引起混乱？

Question

在减少分区数量时，可以使用coalesce，这很好，因为它不会引起混乱，而且似乎可以立即工作（不需要额外的作业阶段）。

有时我想做相反的事情，但重新分区会导致混乱。我想几个月前，我实际上通过使用balanceSlack=1.0的CoalescedRDD实现了这一点-所以会发生的是，它会分割一个分区，从而得到所有分区都位于同一节点上的分区位置（如此小的网络IO）。

这种功能在Hadoop中是自动的，只需调整拆分大小。除非减少分区数量，否则它在Spark中似乎无法以这种方式工作。我认为解决方案可能是编写一个自定义分区程序以及一个自定义RDD，我们在其中定义了getPreferredLocations...但我认为这是一件如此简单和常见的事情，肯定有一种直接的方法可以做到吗？

尝试过的事情：

<代码>。在我的SparkConf上设置（“spark.default.parallelism”，partitions），并且在阅读拼花的上下文中，我尝试了sqlContext。sql（“set spark.sql.shuffle.partitions=…”，它在1.0.0上会导致一个错误，并不是我真正想要的，我希望分区号在所有类型的作业中都能更改，而不仅仅是洗牌。

Answer 1

正如你所知，pyspark使用某种“懒惰”的跑步方式。它只会在有一些操作要执行时进行计算（例如“df.count（）”或“df.show（）”。因此，您可以在这些操作之间定义一个无序分区。

你可以写：

sparkSession.sqlContext().sql("set spark.sql.shuffle.partitions=100")
# you spark code here with some transformation and at least one action
df = df.withColumn("sum", sum(df.A).over(your_window_function))
df.count() # your action

df = df.filter(df.B <10)
df = df.count()   

sparkSession.sqlContext().sql("set spark.sql.shuffle.partitions=10")
# you reduce the number of partition because you know you will have a lot 
# less data
df = df.withColumn("max", max(df.A).over(your_other_window_function))
df.count() # your action

Answer 2

我不太明白你的意思。您的意思是现在有5个分区，但在下一次操作之后，您希望将数据分发到10个分区？因为有10个，但仍然使用5没有多大意义……向新分区发送数据的过程必须在某个时候发生。

在执行合并时，可以去掉未指定的分区，例如：如果最初有100个分区，但在reduceByKey之后有10个分区（因为只有10个键），可以设置合并。

如果您想让流程走另一条路，可以强制进行某种分区：

[RDD].partitionBy(new HashPartitioner(100))

我不确定你在找什么，但希望如此。

Answer 3

注意这个空间

https://issues.apache.org/jira/browse/SPARK-5997

这种非常简单的显而易见的特性最终会被实现——我想就在他们完成了数据集中所有不必要的特性之后。