Spark使用saveAsTextFile如何指定文件扩展名并输出CSV格式

你用的saveAsTextFile是Spark的文本写出接口，本身没有提供自定义文件后缀的参数，你可以根据你的数据类型选择以下两种方案实现CSV后缀的输出：

方案1：使用Spark原生CSV写出接口（优先推荐） #

如果你的records是DataFrame/DataSet类型，直接调用Spark内置的CSV写出方法即可，生成的文件默认自带.csv后缀，还支持自定义表头、分隔符、编码等CSV相关配置，示例代码：

// Scala 示例
records.repartition(partitionNum)
  .write
  .option("header", "true") // 是否写入表头，不需要可以去掉
  .option("delimiter", ",") // 自定义分隔符，默认就是逗号
  .csv(path)

如果是PySpark用法基本一致：

# PySpark 示例
records.repartition(partitionNum) \
  .write \
  .option("header", "true") \
  .csv(path)

方案2：RDD类型使用saveAsTextFile的兼容方案 #

如果你的records是RDD类型，必须用类文本写出的方式，可以写完之后批量重命名文件，或者自定义OutputFormat实现后缀拼接：

方法A：写出后批量重命名 #

通过Hadoop FileSystem API遍历输出目录，给所有part-开头的文件加上.csv后缀：

import org.apache.hadoop.fs.{FileSystem, Path}

val sc = records.sparkContext
val outputPath = new Path(path)
val fs = FileSystem.get(sc.hadoopConfiguration)

fs.listStatus(outputPath).filter(_.getPath.getName.startsWith("part-")).foreach { fileStatus =>
  val oldPath = fileStatus.getPath
  val newPath = new Path(outputPath, s"${oldPath.getName}.csv")
  fs.rename(oldPath, newPath)
}

方法B：自定义OutputFormat #

继承MultipleTextOutputFormat自定义文件名生成逻辑，写出时直接生成带后缀的文件：

import org.apache.hadoop.mapred.lib.MultipleTextOutputFormat
import org.apache.hadoop.io.NullWritable

// 自定义带csv后缀的输出格式
class CsvSuffixOutputFormat extends MultipleTextOutputFormat[NullWritable, String] {
  override def generateFileNameForKeyValue(key: NullWritable, value: String, name: String): String = {
    s"$name.csv"
  }
}

// 写出调用
records.map(line => (NullWritable.get(), line))
  .repartition(partitionNum)
  .saveAsHadoopFile(
    path,
    classOf[NullWritable],
    classOf[String],
    classOf[CsvSuffixOutputFormat]
  )

内容的提问来源于stack exchange，提问作者Rajashree Gr