Spark中的groupByKey是否保留原始顺序？答案

【问题标题】：Does groupByKey in Spark preserve the original order?Spark中的groupByKey是否保留原始顺序？
【发布时间】：2014-06-13 13:35:43
【问题描述】：

在 Spark 中，groupByKey 函数将 (K,V) 对 RDD 转换为 (K,Iterable<V>) 对 RDD。

但是，这个功能稳定吗？即迭代中的顺序是否保留了原始顺序？

例如，如果我最初读取的文件格式为：

K1;V11
K2;V21
K1;V12

我对@987654325@ 的可迭代可能像(V12, V11)（因此不保留原始顺序）或者只能是(V11, V12)（因此保留原来的顺序）？

【问题讨论】：

你本质上是在问它是否是一个稳定的排序？
是的！没错，我问的是稳定性。

标签： scala apache-spark

【解决方案1】：

不，订单不会被保留。 spark-shell 中的示例：

scala> sc.parallelize(Seq(0->1, 0->2), 2).groupByKey.collect
res0: Array[(Int, Iterable[Int])] = Array((0,ArrayBuffer(2, 1)))

顺序取决于时间，因此在运行之间可能会有所不同。（我在下一次运行时得到了相反的顺序。）

这里发生了什么？ groupByKey 通过使用 HashPartitioner 对 RDD 重新分区来工作，以便一个键的所有值都在同一个分区中结束。然后它在每个分区上本地执行聚合。

重新分区也称为“shuffle”，因为 RDD 的行在节点之间重新分配。 shuffle 文件是从其他节点并行提取的。新分区是按照它们到达的顺序从这些部分构建的。来自最慢源的数据将位于新分区的末尾，以及groupByKey 中列表的末尾。

（从worker本身拉取数据当然是最快的。由于这里不涉及网络传输，所以这些数据是同步拉取的，因此是按顺序到达的。（至少看起来如此。）所以复制我的实验，您至少需要 2 个 Spark 工作人员。）

来源：http://apache-spark-user-list.1001560.n3.nabble.com/Is-shuffle-quot-stable-quot-td7628.html

【讨论】：

我仍然遇到的问题是，我觉得并行化正在使这些对乱序，而 groupByKey 仍然稳定，也许还有 parallelize 保留顺序
sc.parallelize(Seq(0->1, 0->2), 2).collect 将始终返回 0->1, 0->2。
好吧，IDT 意味着什么，因为延迟执行 IDT 并行化实际上会做任何事情，直到你执行一个动作
你说你觉得parallelize 是“把配对弄乱了”。我是说绝对不会那样做。你能举一个例子来说明它在哪里构建一个乱序行的 RDD？

【解决方案2】：

Spark（和其他 map reduce 框架）通过分区对数据进行排序，然后合并。由于合并排序是一个稳定的操作，我猜结果是稳定的。在查看更多源代码后，我发现如果 spark.shuffle.spill 为 true，它使用外部排序，在这种情况下合并排序，这是稳定的。如果允许溢出到磁盘，我不能 100% 确定它会做什么。

来源：

private val externalSorting = SparkEnv.get.conf.getBoolean("spark.shuffle.spill", true)

分区也是一种稳定的操作，因为它不会重新排序

【讨论】：

问题不在于排序，而在于是否保留了原始顺序（可能未排序）。
@TravisBrown 这就是稳定排序的意思，“保留原始顺序”，我承认我不是 100% 确定它一直使用合并排序，但它可以肯定；从源头上看起来是这样
对，但是这里没有排序——只是通过键对数据进行分区。
@TravisBrown 数据由一个稳定的键分区，因为它只是线性地遍历它们，然后对分区到同一个“reducer”的数据执行合并
你怎么知道分区（或“洗牌”）是稳定的？它肯定会重新排序。试试sc.parallelize(Seq(1->1, 2->2, 3->3), 3).partitionBy(new HashPartitioner(3)).collect。