Spark：增加分区数量而不引起洗牌？

Question

当减少分区数量时，可以使用coalesce，这非常棒，因为它不会导致随机播放并且似乎可以立即工作（不需要额外的工作阶段）。

有时我想反其道而行之，但repartition 会引发洗牌。我想几个月前我实际上是通过使用CoalescedRDD 和balanceSlack = 1.0 来完成这项工作的——所以会发生什么情况是它会拆分一个分区，以便生成的分区位于同一个节点上（如此小的网络IO）。

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

尝试过的事情：

.set("spark.default.parallelism", partitions) 在我的SparkConf 上，并且在阅读镶木地板的情况下我尝试过sqlContext.sql("set spark.sql.shuffle.partitions= ...，它在 1.0.0 上会导致错误并且不是我想要的，我希望分区号改变所有类型的工作，而不仅仅是洗牌。

Answer 1

注意这个空间

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

这种非常简单明显的功能最终会实现——我猜他们在完成Datasets 中所有不必要的功能之后。

Answer 2

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

在执行coalesce 时，您可以摆脱未使用的分区，例如：如果您最初有 100 个，但在 reduceByKey 之后您得到了 10 个（因为那里只有 10 个键），您可以设置 coalesce。

如果您希望进程以另一种方式进行，您可以强制进行某种分区：

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

我不确定这就是你要找的东西，但希望如此。

Answer 3

如您所知，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