Spark/Scala：根据迭代索引动态命名文件写入DataFrame

刚遇到这个需求的时候我也踩过坑，其实Scala里有好几种简单的方法来实现动态文件名，结合Spark的特性给你梳理一下：

一、最简洁的方式：Scala字符串插值 #

Scala的字符串插值（以s开头的字符串）可以直接在字符串里引用变量，写法非常直观，完全满足你的需求：

for (idx <- 1 to 3) {
  // 这里写生成依赖idx的DataFrame的逻辑
  // ...
  
  // 用s插值动态生成文件名
  val outputPath = s"/temp/path/file${idx}.csv"
  df.coalesce(1).write.csv(outputPath)
}

这样循环时就会依次生成file1.csv、file2.csv、file3.csv对应的路径啦。

二、灵活格式化：使用String.format() #

如果需要更复杂的文件名格式（比如固定位数的索引，比如file001.csv这种），可以用String.format()，它支持类似Java的格式化语法：

for (idx <- 1 to 3) {
  // 生成df的逻辑...
  
  // 普通数字格式
  val outputPath = String.format("/temp/path/file%d.csv", idx)
  
  // 如果需要补零生成file001.csv这种格式，用%03d表示3位数字，不足补零
  // val outputPath = String.format("/temp/path/file%03d.csv", idx)
  
  df.coalesce(1).write.csv(outputPath)
}

重要注意事项：Spark写CSV的目录特性 #

这里要提醒你一个容易踩的坑：Spark的write.csv()方法并不会直接生成你指定的单个CSV文件，而是会创建一个以你指定路径命名的目录，然后在目录里生成part-00000-...开头的CSV文件（哪怕你用了coalesce(1)也只是保证目录里只有一个part文件）。

如果需要直接得到file1.csv这样的单个文件，需要额外做重命名操作，比如用Hadoop的FileSystem API来实现：

import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.sql.SparkSession

// 先获取SparkSession和FileSystem实例
val spark = SparkSession.builder().getOrCreate()
val fs = FileSystem.get(spark.sparkContext.hadoopConfiguration)

for (idx <- 1 to 3) {
  // 生成df的逻辑...
  
  // 定义临时目录和最终文件路径
  val tempDir = s"/temp/path/temp_file${idx}"
  val finalCsvPath = s"/temp/path/file${idx}.csv"
  
  // 先把DataFrame写入临时目录
  df.coalesce(1).write.mode("overwrite").csv(tempDir)
  
  // 找到临时目录里的part文件
  val partFiles = fs.listStatus(new Path(tempDir))
    .filter(status => status.getPath.getName.startsWith("part-"))
  
  if (partFiles.nonEmpty) {
    // 重命名part文件到最终路径
    val sourcePath = partFiles(0).getPath
    fs.rename(sourcePath, new Path(finalCsvPath))
  }
  
  // 删除临时目录
  fs.delete(new Path(tempDir), true)
}

这段代码会先把数据写到临时目录，找到里面的part文件后重命名为你想要的文件名，最后删掉临时目录，完美解决单个文件的需求。

内容的提问来源于stack exchange，提问作者Guanghua Shu