在Apache Spark中混合与非混合合并

Question

当以下转换在将RDD写入文件之前执行时，它们之间有什么区别？

1. 聚结（1，洗牌=true）
2. 合并（1，洗牌=假）

代码示例：

val input = sc.textFile(inputFile)
val filtered = input.filter(doSomeFiltering)
val mapped = filtered.map(doSomeMapping)

mapped.coalesce(1, shuffle = true).saveAsTextFile(outputFile)
vs
mapped.coalesce(1, shuffle = false).saveAsTextFile(outputFile)

它与collect（）相比如何？我完全知道Spark save方法将以HDFS风格的结构存储它，但我更感兴趣的是collect（）和shuffled/non shuffled coalesce（）的数据分区方面。

Answer 1

coalesce（n， schffle=true）也相当于re分区（n）可能会对作业的执行产生相当大的影响，这取决于父RDD中的映射或任何其他处理登录。

一般来说，如果父分区中的数据分布均匀，并且没有大幅减少分区数，那么在使用coalesce时应该避免使用shuffle。

然而，在您的情况下，这是分区数量的大幅减少，并且符合文档要求

但是，如果您正在进行剧烈合并，例如to numPartitions=1，这可能会导致您的计算发生在比您希望的更少的节点上（例如，numPartitions=1的情况下是一个节点）。为了避免这种情况，可以传递shuffle=true。这将添加一个洗牌步骤，但意味着当前的上游分区将并行执行（无论当前分区是什么）

考虑到这一点，现在你需要正确地评估和选择

• 洗牌潜在的大量数据，但在父分区中并行进行计算

例如，考虑以下片段，这些片段与你可能拥有的实际逻辑相去甚远，但会给你一个正在发生的事情的视角

// fast
sc.parallelize(0 to 1000000, 10)
  .mapPartitions(it => {Thread.sleep(5000); it.map(_.toString)})
  .coalesce(1, shuffle = true)
  .toDF.write.text("shuffleTrue")

// slow
sc.parallelize(0 to 1000000, 10)
  .mapPartitions(it => {Thread.sleep(5000); it.map(_.toString)})
  .coalesce(1, shuffle = false)
  .toDF.write.text("shuffleFalse")

在我的集群上，使用shuffle=true显示10个任务的总时间约为5秒，在每个父分区上并行执行计算逻辑。而另一个使用shuffle=false的执行器在一个任务中，大约有50秒的时间在一个执行器上完成所有计算。

Answer 2

通过查看Spark 2.3.1的合并文档，https://spark.apache.org/docs/2.3.1/api/java/org/apache/spark/rdd/RDD.html#coalesce-int布尔scala。选择scala。数学订购-

当您将分区数量减少到1以避免计算发生在比您喜欢的更少的节点上时，添加Shuffle=true看起来更方便。这将增加一个洗牌步骤，但意味着当前上游分区将并行执行。

Answer 3

Shuffle=true和Shuffle=false在结果输出中不会有任何实际差异，因为它们都将下降到单个分区。然而，当您将其设置为true时，您将执行没有任何用处的洗牌。使用Shuffle=true，输出在分区之间均匀分布（如果你愿意，你也可以增加分区的#），但是因为你的目标是1个分区，所以不管怎样，一切都结束在一个分区中。

至于与collect（）的比较，区别在于所有数据都存储在单个执行器上，而不是存储在驱动程序上。