重新分区后引发非确定性结果答案

【问题标题】：Spark non-deterministic results after repartition重新分区后引发非确定性结果
【发布时间】：2020-08-14 21:54:08
【问题描述】：

有什么方法可以在不排序的情况下从数据帧重新分区中获得确定性结果？在下面的代码中，我在执行相同的操作时得到了不同的结果。

from pyspark.sql.functions import rand, randn
from pyspark.sql import SQLContext
sqlContext = SQLContext(sc)
df = sqlContext.range(0, 100000)

# repartition dataframe to 5 partitions
df2 = df.repartition(5).persist()
df2.head(5)

Out[1]: [Row(id=5324), Row(id=5389), Row(id=6209), Row(id=7640), Row(id=8090)]

df2.unpersist()
df3 = df.repartition(5).persist()
df3.head(5)

Out[2]: [Row(id=1019), Row(id=652), Row(id=2287), Row(id=470), Row(id=1348)]

Spark 版本 - 2.4.5

【问题讨论】：

那会慢一些。做orderby。
感谢您的评论。我正在寻找不排序的方法（也编辑了帖子）。不确定这是否可能。
您使用的是什么 Spark 版本？
@mazaneicha 我使用的是 Spark 2.4.5

标签： apache-spark pyspark

【解决方案1】：

此non deterministic 行为是expected。方法如下...

.repartition(num) 在函数内部没有传递任何列时执行round-robin 重新分区。这并不能保证特定行将始终位于特定分区中。
.head(n) 返回数据帧第一个分区的前 n 行。

如果要下单，需要使用orderBy！

【讨论】：

【解决方案2】：

根据JIRA，重新分区（默认情况下）涉及本地排序，并且是完全确定的。来自 PR 说明：

在这个 PR 中，我们建议……在执行本地排序之前分区，在我们使输入行排序确定后，从行到分区的函数也是完全确定的。

这种方法的缺点是插入了额外的本地排序， repartition() 的性能会下降，所以我们添加一个新配置命名为spark.sql.execution.sortBeforeRepartition 来控制是否此补丁已应用。该补丁默认启用为默认情况下是安全的，但用户可以选择手动关闭它以避免性能回归。

另一方面，head(n) 不是（除非您应用 orderBy 再次将数据集重新分区到一个分区），但这不是您关心的问题吗？

【讨论】：

感谢您的回答。您的意思是“默认情况下，重新分区是确定性的，而 head(n) 是非确定性的”。那么我如何确定某些东西是否是确定性的呢？为了提供更多上下文，我使用重新分区的数据框进行模型训练，并且我从模型预测中得到了不确定的结果。但是，当我不重新分区时，我会在不同的运行中看到相同的结果。
抱歉，在你的情况下，df 实际上是PythonRDD，我认为我提到的 JIRA 不适用，因为 rdds 遵循不同的代码路径。
即当我使用 spark-shell 尝试您的确切代码时，每次都会得到相同的（确定性）输出。