相关疑难解决方法(0)

在Spark中使用Scala时,每当我使用结果转储结果时saveAsTextFile,它似乎将输出分成多个部分.我只是将一个参数(路径)传递给它.

val year = sc.textFile("apat63_99.txt").map(_.split(",")(1)).flatMap(_.split(",")).map((_,1)).reduceByKey((_+_)).map(_.swap)
year.saveAsTextFile("year")

1. 输出数量是否与其使用的减速器数量相对应？
2. 这是否意味着输出被压缩了？
3. 我知道我可以使用bash将输出组合在一起,但是有一个选项可以将输出存储在单个文本文件中,而不会拆分吗？我查看了API文档,但对此并没有太多说明.

我正在处理一个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

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

我的具体代码如下所示:

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

谁能提出建议？谢谢.

(我也注意到这个线程讨论了类似的问题,但到目前为止还没有一个好的解决方案.如果这种现象真的是由重新分区操作引起的,那么任何人都可以帮助我有效地将数据帧转换为单个JSON文件而不进行转换进入RDD,同时保持排序顺序？谢谢)

方案:

非常感谢@manos @eliasah和@pkrishna的帮助.在阅读了你的评论之后,我曾考虑过使用coalesce,但在调查了它的性能后,我放弃了这个想法.

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

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

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