scala - spark中的saveAsTextFile方法

Question

在我的项目中，我有三个输入文件，并将文件名设为 args(0) 到 args(2)，我还有一个输出文件名作为 args(3)，在源代码中，我使用

val sc = new SparkContext()
var log = sc.textFile(args(0))
for(i <- 1 until args.size - 1) log = log.union(sc.textFile(args(i)))

我对日志不做任何处理，但使用以下命令将其保存为文本文件

log.coalesce(1, true).saveAsTextFile(args(args.size - 1))

但它仍然以part-00000、part-00001、part-00002保存到3个文件，那么有什么办法可以将三个输入文件保存到输出文件中？

Answer 1

拥有多个输出文件是 Hadoop 或 Spark 等多机集群的标准行为。输出文件的数量取决于 reducer 的数量。

如何在 Hadoop 中“解决”它:
merge output files after reduce phase

如何在 Spark 中“解决”:
how to make saveAsTextFile NOT split output into multiple file?

你也可以在这里获得一个很好的信息:
http://apache-spark-user-list.1001560.n3.nabble.com/How-to-make-Spark-merge-the-output-file-td322.html

所以，你是对的 coalesce(1,true) .然而，这是非常低效的。有趣的是(正如@climbage 在他的评论中提到的)如果你在本地运行你的代码就可以工作。

您可能会尝试先读取文件，然后保存输出。

...
val sc = new SparkContext()
var str = new String("")
for(i <- 0 until args.size - 1){
   val file = sc.textFile(args(i))       
   file.foreach(line => str+= line)
}
//and now you might save the content
str.coalesce(1, true).saveAsTextFile("out")

注意:此代码也非常低效，仅适用于小文件!!!你需要想出一个更好的代码。我不会尝试减少文件数量，而是处理多个输出文件。