在Apache Spark中指定输出文件名

Question

我有一个MapReduce的任务，想把它迁移到PySpark上。请问有没有办法定义输出文件的名字，而不是得到那种part-xxxxx的格式呢？

在MapReduce中，我是用org.apache.hadoop.mapred.lib.MultipleTextOutputFormat这个类来实现的。

顺便说一下，我确实试过saveAsTextFile()这个方法。例如：

lines = sc.textFile(filesToProcessStr)
counts = lines.flatMap(lambda x: re.split('[\s&]', x.strip()))\
.saveAsTextFile("/user/itsjeevs/mymr-output")

这样做会生成相同的part-0000文件。

[13:46:25] [spark] $ hadoop fs -ls /user/itsjeevs/mymr-output/
Found 3 items
-rw-r-----   2 itsjeevs itsjeevs          0 2014-08-13 13:46 /user/itsjeevs/mymr-output/_SUCCESS
-rw-r--r--   2 itsjeevs itsjeevs  101819636 2014-08-13 13:46 /user/itsjeevs/mymr-output/part-00000
-rw-r--r--   2 itsjeevs itsjeevs   17682682 2014-08-13 13:46 /user/itsjeevs/mymr-output/part-00001

补充说明

最近我读到了一篇文章，这对Spark用户来说会简单很多。

Answer 1

你的输出文件会像这样：

myfilename-r-00000
myfilename-r-00001

        SparkConf sparkConf = new SparkConf().setAppName("WCSYNC-FileCompressor-ClusterSaver");
        SparkContext sc = new SparkContext(sparkConf);
            JavaSparkContext context = new JavaSparkContext(sc)
context.hadoopConfiguration().set("mapreduce.output.basename", "myfilename");




saveAsNewAPIHadoopFile(outputpath,
                                Text.class,
                                Text.class,
                                TextOutputFormat.class,
                                context.hadoopConfiguration());

Answer 2

这是我在Python中能做到的，离被接受的答案最近的结果：

def saveAsSingleJsonFile(df, path, filename):
    rdd = df.toJSON()
    rdd.map(lambda x: (None, x)).saveAsNewAPIHadoopFile(
        f"{path}/{filename}",
        "org.apache.hadoop.mapreduce.lib.output.TextOutputFormat",
        keyClass="org.apache.hadoop.io.NullWritable",
        valueClass="org.apache.hadoop.io.Text"
    )

Answer 3

Spark其实在后台也使用了Hadoop，所以你可能能实现你想要的功能。下面是saveAsTextFile的实现方式：

def saveAsTextFile(path: String) {
  this.map(x => (NullWritable.get(), new Text(x.toString)))
    .saveAsHadoopFile[TextOutputFormat[NullWritable, Text]](path)
}

你可以给saveAsHadoopFile传入一个自定义的OutputFormat。不过我不知道怎么在Python中做到这一点，抱歉没能给出完整的答案。