如何将 RDD 数据保存到 json 文件中，而不是文件夹中答案

【问题标题】：How to save RDD data into json files, not folders如何将 RDD 数据保存到 json 文件中，而不是文件夹中
【发布时间】：2017-03-27 10:18:12
【问题描述】：

我正在接收我想保存在 S3 中的流数据myDStream (DStream[String])以防万一）。

以下代码运行良好，但它保存名称为jsonFile-19-45-46.json 的文件夹，然后在文件夹中保存文件_SUCCESS 和part-00000。

是否可以将每个 RDD[String]（这些是 JSON 字符串）数据保存到 JSON 文件，而不是文件夹中？我以为repartition(1) 一定要搞这个把戏，但它没有。

    myDStream.foreachRDD { rdd => 
       // datetimeString = ....
       rdd.repartition(1).saveAsTextFile("s3n://mybucket/keys/jsonFile-"+datetimeString+".json")
    }

【问题讨论】：

请参阅how to make saveAsTextFile NOT split output into multiple file?，了解它为何如此运作以及如何处理。
@hoyland：我检查了这个线程，建议使用coalesce(1,true)。我尝试了这个解决方案，但我仍然得到带有 SUCCESS 和 part00000 的文件夹，而不是 json 文件。就我而言，我知道每个 RDD 的大小会相对较小。

标签： scala apache-spark spark-streaming

【解决方案1】：

AFAIK 没有将其保存为文件的选项。因为它是一个分布式处理框架，在单个文件上写入而不是每个分区在指定路径中写入自己的文件并不是一个好习惯。

我们只能传递我们想要保存数据的输出目录。 OutputWriter 将在带有part- 文件名前缀的指定路径内创建文件（取决于分区）。

【讨论】：

是的，我明白了。对于我的应用程序，我决定使用rdd.collect.mkString("\n")，然后使用 Amazon SDK 将字符串作为文件上传到 S3。它可以满足我的需求。

【解决方案2】：

作为rdd.collect.mkString("\n") 的替代方案，您可以使用hadoop 文件系统库通过将part-00000 文件移动到它的位置来清理输出。以下代码在本地文件系统和 HDFS 上完美运行，但我无法使用 S3 对其进行测试：

val outputPath = "path/to/some/file.json"
rdd.saveAsTextFile(outputPath + "-tmp")

import org.apache.hadoop.fs.Path
val fs = org.apache.hadoop.fs.FileSystem.get(spark.sparkContext.hadoopConfiguration)
fs.rename(new Path(outputPath + "-tmp/part-00000"), new Path(outputPath))
fs.delete(new Path(outputPath  + "-tmp"), true)

【讨论】：

【解决方案3】：

对于 JAVA，我实现了这个。希望对您有所帮助：

    val fs = FileSystem.get(spark.sparkContext().hadoopConfiguration());
    File dir = new File(System.getProperty("user.dir") + "/my.csv/");
    File[] files = dir.listFiles((d, name) -> name.endsWith(".csv"));
    fs.rename(new Path(files[0].toURI()), new Path(System.getProperty("user.dir") + "/csvDirectory/newData.csv"));
    fs.delete(new Path(System.getProperty("user.dir") + "/my.csv/"), true);

【讨论】：