将DataFrame保存为CSV时指定文件名

Question

假设我有一个Spark DF,我想将其保存到磁盘CSV文件.在Spark 2.0.0+中,可以转换DataFrame(DataSet[Rows])为a DataFrameWriter并使用该.csv方法来编写文件.

该功能定义为

def csv(path: String): Unit
    path : the location/folder name and not the file name.

Spark将csv文件存储在指定位置,方法是创建名称为part - *.csv的CSV文件.

有没有办法用指定的文件名而不是部分保存CSV - *.csv？或者可以指定前缀而不是part-r？

代码:

df.coalesce(1).write.csv("sample_path")

电流输出:

sample_path
|
+-- part-r-00000.csv

期望的输出:

sample_path
|
+-- my_file.csv

注意: coalesce函数用于输出单个文件,执行程序有足够的内存来收集DF而没有内存错误.

Answer 1

它不可能直接在Spark中完成 save

Spark使用Hadoop文件格式,这需要对数据进行分区 - 这就是您拥有part-文件的原因.您可以在处理后轻松更改文件名,就像在这个问题中一样

在Scala中,它看起来像:

import org.apache.hadoop.fs._;
val fs = FileSystem.get(sc.hadoopConfiguration);
val file = fs.globStatus(new Path("path/file.csv/part*"))(0).getPath().getName();

fs.rename(new Path("csvDirectory/" + file), new Path("mydata.csv"));
fs.delete(new Path("mydata.csv-temp"), true);

要不就:

import org.apache.hadoop.fs._;
val fs = FileSystem.get(sc.hadoopConfiguration());
fs.rename(new Path("csvDirectory/data.csv/part-0000"), new Path("csvDirectory/newData.csv"));

编辑:如评论中所述,您也可以编写自己的OutputFormat,请参阅文档以获取有关此方法设置文件名的信息