apache-spark - 排序后未对 Spark 数据框进行排序

Question

我正在处理一个 JSON 文件以使用 Spark(版本 1.6.1)生成两个 JSON 文件。输入文件的大小约为 30~40G(100M 记录)。生成的文件，大的大约10G~15G(30M记录)，小的大约500M~750M(1.5M记录)。两个结果文件都面临以下问题:

我调用了数据框的“排序”方法，然后执行“重新分区”以将结果合并到一个文件中。然后我检查了生成的文件，发现 在记录被排序的时间间隔内，但整个文件没有全局排序 。例如文件中最后一条记录(第 1.9M 行)的键(由 3 列构成)是“(ou7QDj48c, 014, 075)”，但文件中一条中间记录(第 375K 行)的键是“( pzwzh5vm8, 003, 023)"

pzwzh5vm8 003 023
...
ou7QDj48c 014 075

当我使用相对较小的输入源(输入文件 40 万行)在本地测试代码时，这种情况根本不会发生。

我的具体代码如下所示:

big_json = big_json.sort($"col1", $"col2", $"col3", $"col4")
big_json.repartition(1).write.mode("overwrite").json("filepath")

谁能给个建议？谢谢你。

(我也注意到 this thread讨论了一个类似的问题，但目前还没有很好的解决方案。如果这种现象真的是重新分区操作造成的，谁能帮我有效地将dataframe转换为单个JSON文件而不将其转换为RDD，同时保持排序顺序？谢谢)

解决方案 :

非常感谢@manos @eliasah 和@pkrishna 的帮助。阅读您的评论后，我曾考虑使用合并，但在调查了其性能后，我放弃了这个想法。

最终的解决方案是:对数据帧进行排序并写入JSON，无需任何重新分区或合并。全部工作完成后，调用下面的HDFS命令

hdfs dfs -getmerge /hdfs/file/path/part* ./local.json

这个命令比我想象的要好得多。它既不会占用太多时间也不会占用太多空间，并且给了我一个很好的单个文件。我只是在巨大的结果文件上使用了 head 和 tail，它似乎完全有序。

Answer 1

发生的情况是，在 对 排序 操作之后，您正在 重新分区 。
repartition 随机重组 RDD 中的数据以创建更多或更少的分区并在它们之间进行平衡。这总是在网络上打乱所有数据。

在幕后，它使用 coalesce 和 shuffle 重新分配数据。
这就是您的数据不再排序的原因。

您可以检查 reference 的代码。