在 Spark Streaming 中更改输出文件名答案

【问题标题】：Change output file name in Spark Streaming在 Spark Streaming 中更改输出文件名
【发布时间】：2016-10-24 15:39:11
【问题描述】：

我正在运行一个 Spark 作业，就逻辑而言，它的性能非常好。但是，当我使用 saveAsTextFile 将文件保存在 s3 存储桶中时，我的输出文件的名称格式为 part-00000、part-00001 等。有没有办法改变输出文件名？

谢谢。

【问题讨论】：

Renaming Part Files in Hadoop Map Reduce的可能重复
最好使用 shell 而不是在 Spark 中执行此操作。例如，您可以使用coalesce 将所有内容收集到一个文件中，但它会给内存带来压力——此外，hdfs 的工作方式与常规文件系统略有不同，Spark 总是为每个输出创建不同的目标/文件夹。
难道不是这种...gist.github.com/mlehman/df9546f6be2e362bbad2

标签： hadoop apache-spark spark-streaming spark-dataframe

【解决方案1】：

在 Spark 中，您可以使用 saveAsNewAPIHadoopFile 并在 hadoop 配置中设置 mapreduce.output.basename 参数来更改前缀（只是“part”前缀）

val hadoopConf = new Configuration()
hadoopConf.set("mapreduce.output.basename", "yourPrefix")

yourRDD.map(str => (null, str))
        .saveAsNewAPIHadoopFile(s"$outputPath/$dirName", classOf[NullWritable], classOf[String],
          classOf[TextOutputFormat[NullWritable, String]], hadoopConf)

您的文件将被命名为：yourPrefix-r-00001

在 hadoop 和 Spark 中，您可以在输出中包含多个文件，因为您可以拥有多个 reducer (hadoop) 或多个分区 (spark)。然后你需要保证它们每个的名称都是唯一的，这就是为什么不能覆盖文件名最后部分的序列号的原因。

但如果您想更好地控制文件名，您可以扩展 TextOutputFormat 或 FileOutputFormat 并覆盖 getUniqueFile 方法。 p>

【讨论】：

感谢您的评论。我们可以将文件保存到 s3 存储桶而不是 hdfs 吗？
可以，Amazon S3 与 Hadoop API 兼容。
Hadoop 文件系统实现了多个协议/文件系统，包括 S3：wiki.apache.org/hadoop/AmazonS3。您可以使用几乎透明的任何一种（您只需为每种连接指定具体参数）。如果您认为我的回答对您最初的问题有所帮助，请接受。
你能不能看看这个问题stackoverflow.com/questions/46703623/…
我想现在是getUniqueName。有没有办法可以将行中的字段用作文件名，例如身份证？

【解决方案2】：

[Java 解决方案]

假设你有：

JavaRDD<Text> rows;

并且您想将其写入 customPrefix-r-00000 之类的文件中。

Configuration hadoopConf = new Configuration();
hadoopConf.set("mapreduce.output.basename", "customPrefix");

rows.mapToPair(row -> new Tuple2(null, row)).saveAsNewAPIHadoopFile(outputPath, NullWritable.class, Text.class, TextOutputFormat.class, hadoopConf);

多田！！

【讨论】：

嗨，我尝试设置 basename 但它不起作用：job.getConfiguration().set("mapreduce.output.basename", inputStrategyName); javaPairRDD.saveAsNewAPIHadoopFile(outputPath, new AvroKey().getClass(), NullWritable.class, AvroKeyOutputFormat.class, job.getConfiguration());
作业对象的类别是什么？新建一个该类的配置对象：org.apache.hadoop.conf.Configuration;