如何为DataFrame中的每条记录调用指定方法生成对应文本文件

1 先修复原有写文件逻辑的问题 #

你现有写文件的代码有两处会导致功能异常的问题：

• 创建FileWriter时传入的是无路径的fileName，最终文件会写入程序运行的当前工作目录，不会存到你指定的Path路径下
• 输出流使用完没有关闭，会大概率出现内容丢失、文件为空的问题

修正后的写文件方法参考：

import java.io.{File, FileWriter, PrintWriter}

def writeToFile(path: String, fileName: String, text: String): Unit = {
  // 拼接路径时用File.separator避免操作系统斜杠不兼容问题
  val fileWithAbsolutePath = new File(path + File.separator + fileName)
  var printWriter: PrintWriter = null
  try {
    val fileWriter = new FileWriter(fileWithAbsolutePath)
    printWriter = new PrintWriter(fileWriter)
    printWriter.print(text)
  } finally {
    // 确保流一定关闭
    if (printWriter != null) {
      printWriter.close()
    }
  }
}

如果需要相同文件名追加内容而非覆盖，可以把new FileWriter(fileWithAbsolutePath)改为new FileWriter(fileWithAbsolutePath, true)。

2 遍历DataFrame每一行调用方法 #

直接调用DataFrame的foreach算子即可逐行处理所有记录，从Row中取出对应字段传入写文件方法：

// 假设你的DataFrame变量名为df
df.foreach { row =>
  // 按字段名取出对应的值
  val fileName = row.getAs[String]("fileName")
  val path = row.getAs[String]("Path")
  val text = row.getAs[String]("Text")
  // 调用写文件方法
  writeToFile(path, fileName, text)
}

注意事项 #

• 如果你是在Spark集群上运行任务，需要保证Path是所有Executor节点都能访问的共享存储路径（比如挂载的NAS、HDFS路径等），否则文件会写入各Executor的本地目录，无法在Driver端统一找到
• 如果你的数据量极大，不建议直接逐行写本地文件，IO开销会非常高，可以考虑先按fileName分组再批量写入

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