我正在尝试找到一种将 Spark Job 的结果保存为 csv 文件的有效方法。我正在使用 Spark 和 Hadoop，到目前为止，我的所有文件都保存为 part-00000。
任何想法如何使我的火花保存到具有指定文件名的文件？
【问题讨论】：
标签： file csv hadoop apache-spark distributed-computing
由于 Spark 使用 Hadoop File System API 将数据写入文件，这是不可避免的。如果你这样做了
rdd.saveAsTextFile("foo")
它将保存为“foo/part-XXXXX”，在您尝试保存的 RDD 中的每个分区中都有一个 part-* 文件。将 RDD 中的每个分区写入一个单独的文件的原因是为了容错。如果写入第三个分区（即part-00002）的任务失败，Spark 只需重新运行任务并覆盖部分写入/损坏的part-00002，对其他部分没有影响。如果它们都写入同一个文件，那么恢复单个任务失败的难度要大得多。
part-XXXXX 文件通常不是问题，如果您打算在基于 Spark / Hadoop 的框架中再次使用它，因为它们都使用 HDFS API，如果您要求它们读取“foo”，它们都会读取foo 中的所有part-XXXXX 文件也是如此。
【讨论】：
hdfs dfs -getmerge <src-directory> <dst-file>
repartition(1) 将数据重新分区到单个输出文件中。这只适用于小数据，例如，当您想将 CSV 交给分析师以在 Excel 中查看时。
coalesce(1)
我建议这样做（Java 示例）：
theRddToPrint.coalesce(1, true).saveAsTextFile(textFileName);
FileSystem fs = anyUtilClass.getHadoopFileSystem(rootFolder);
FileUtil.copyMerge(
fs, new Path(textFileName),
fs, new Path(textFileNameDestiny),
true, fs.getConf(), null);
【讨论】：
使用 Spark SQL，我们可以在一个班轮中做到这一点
//implicits for magic functions like .toDf
import spark.implicits._
val df = Seq(
("first", 2.0),
("choose", 7.0),
("test", 1.5)
).toDF("name", "vals")
//write DataFrame/DataSet to external storage
df.write
.format("csv")
.save("csv/file/location")
【讨论】：
我有一个想法，但没有准备好代码 sn-p。 Spark 在内部（顾名思义）使用 Hadoop 输出格式。 （以及从 HDFS 读取时的InputFormat）。
在hadoop的FileOutputFormat中有一个受保护的成员setOutputFormat，你可以从继承的类中调用它来设置其他的基名。
【讨论】：
这并不是一个真正干净的解决方案，但在 foreachRDD() 中你基本上可以做任何你喜欢的事情，也可以创建一个新文件。
在我的解决方案中，这就是我所做的：我将输出保存在 HDFS 上（出于容错原因），并在 foreachRDD 内我还在本地文件夹中创建了一个带有统计信息的 TSV 文件。
如果你需要的话，我认为你也可以这样做。
http://spark.apache.org/docs/0.9.1/streaming-programming-guide.html#output-operations
【讨论】：