Spark df.collect() 导致随机不一致的输出答案

【问题标题】：Spark df.collect() results in random inconsistent outputSpark df.collect() 导致随机不一致的输出
【发布时间】：2016-01-28 22:39:27
【问题描述】：

没有太深入细节，我想知道是否有人遇到过类似的奇怪行为。

我在 scala spark-shell, spark 1.4.1 中运行所有这些。

我有一个名为“数据”的 spark df。基本上这些是使用 sql.context 读入 spark 的 parquet 文件。我对这个数据集进行了一些转换，包括过滤器、groupBy、排序、计数……没有什么太花哨的，所有这些确定性，没有任何随机性。通过这种方式，我正在创建一些派生的 df 的子集_1 和子集_2。最后，我正在运行以下类型的计算：

data.join(subset_1,"key_A").withColumnRenamed("count","count_1").join(subset_2,"key_A").filter("Feature LIKE 'inactive'").groupBy($"key_A",$"key_B").count.withColumnRenamed("count","count_2").groupBy($"key_A").count.withColumnRenamed("count","count_3").groupBy($"count_3").count.collect()

从语法的角度来看，这个计算运行得“很好”。但是，在此查询的不同运行中，我得到不同的结果。例如：

res82: Array[org.apache.spark.sql.Row] = Array([31,3], [32,2], [34,1], [35,1], [38,1], [42,1], [44,1], [52,1], [61,2], [81,1], [1,4933], [2,2361], [3,924], [4,441], [5,220], [6,130], [7,80], [8,59], [9,36], [10,24], [11,13], [12,12], [13,7], [14,7], [15,11], [16,6], [17,4], [18,6], [19,3], [20,5], [21,6], [22,3], [24,1], [25,1], [26,2], [27,2], [28,1], [29,1], [30,3])

和

res81: Array[org.apache.spark.sql.Row] = Array([32,3], [35,3], [43,1], [46,2], [52,1], [122,1], [145,1], [165,1], [1,3515], [2,1887], [3,836], [4,381], [5,238], [6,136], [7,84], [8,51], [9,39], [10,28], [11,28], [12,13], [13,7], [14,13], [15,8], [16,10], [17,8], [18,6], [19,4], [20,2], [21,4], [22,3], [23,4], [24,1], [25,2], [26,1], [28,3], [29,1], [30,2])

同样，相同的数据，相同的代码，在我所做的事情中没有包含随机性，仍然是：随机结果。

非常感谢任何想法。

【问题讨论】：

标签： scala apache-spark

【解决方案1】：

好的，我遇到的“问题”与以下有关：

Spark sort by key and then group by to get ordered iterable?

基本上，必须非常小心 sort 和 groupBy 的组合。

示例：假设您有一个数据框df，其中包含person、status 和date 列。假设您想获取某人的最新状态。人们可能会考虑做类似的事情：

df.sort($"date").groupBy($"person").agg(sql.functions.last($"status"))

不幸的是，如果你 .collect 结果，并且你尝试了很多次，你会意识到结果可能不同（我猜，如果 df 底层的数据恰好由一个分区组成，它们只会是相同的） .这是因为 .sort 是在分区上本地完成的，并且在 groupBy 期间将分区放在一起的方式决不能保证在组内保持 $"date" 的“全局顺序”。

【讨论】：