Spark Streaming：如何将DStream或RDD内容追加至已有输出文件

阿华AIGC实验室

2026-5-13

解决Spark Streaming追加写入文本文件的问题

我明白你的需求啦——你正在开发一个Spark Streaming的单词计数示例，监听TCP Socket接收文本数据，想要把每个非空DStream的内容追加到已有的文本文件里，但目前用saveAsTextFile每次都会覆盖掉旧文件，对吧？

为什么`saveAsTextFile`会覆盖文件？

Spark的saveAsTextFile API本质是为分布式输出设计的，它会在指定路径下生成多个分区文件（比如part-00000），如果指定的是单个文件名，Spark会把它当成目录名来处理，每次调用都会清空该目录并重新写入，这就导致了旧数据被覆盖的问题。

解决方案：用Hadoop FileSystem API实现追加写入

因为Spark底层依赖Hadoop生态，我们可以直接使用Hadoop的FileSystem API来实现追加写入，这种方式适合分布式场景，性能也更好。下面是修改后的代码：

import org.apache.spark.{SparkContext, SparkConf}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.storage.StorageLevel
import org.apache.log4j.{Level, Logger}
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.hadoop.io.IOUtils
import java.io.OutputStreamWriter

Logger.getRootLogger.setLevel(Level.WARN)
val ssc = new StreamingContext(sc, Seconds(2))
val lines = ssc.socketTextStream("localhost", 9999, StorageLevel.MEMORY_AND_DISK_SER)

lines.foreachRDD { rdd =>
  if (!rdd.isEmpty) {
    // 获取Hadoop文件系统实例
    val fs = FileSystem.get(sc.hadoopConfiguration)
    val outputFilePath = new Path("/stream_test/testLine.txt")
    
    // 先检查文件是否存在，不存在则创建空文件
    if (!fs.exists(outputFilePath)) {
      fs.create(outputFilePath).close()
    }
    
    // 对每个分区的数据进行追加写入，减少流的打开关闭开销
    rdd.foreachPartition { partitionIter =>
      // 打开追加模式的输出流
      val outputStream = fs.append(outputFilePath)
      val writer = new OutputStreamWriter(outputStream, "UTF-8")
      
      try {
        // 遍历分区内的每条记录，写入文件并换行
        partitionIter.foreach { line =>
          writer.write(line)
          writer.write("\n")
        }
      } finally {
        // 确保流被关闭，避免资源泄漏
        IOUtils.closeStream(writer)
      }
    }
  }
}

ssc.start()
ssc.awaitTermination()

代码要点说明

FileSystem.append()：这是实现追加的核心方法，它会打开一个支持追加的输出流，不管是HDFS还是本地文件系统都能兼容。
foreachPartition：相比直接用foreach，每个分区只打开一次流，大大减少了IO操作的开销，更适合分布式场景。
文件存在性检查：如果目标文件不存在，append方法会报错，所以我们先判断并创建空文件。

备选方案：Driver端收集数据追加（仅适合小数据量）

如果你的数据量很小，可以把RDD的数据收集到Driver端，用普通的Java/Scala文件IO实现追加，但要注意这种方式会把所有数据拉到Driver节点，数据量大时容易出现内存溢出：

lines.foreachRDD { rdd =>
  if (!rdd.isEmpty) {
    val dataList = rdd.collect()
    import java.io._
    // 第二个参数true表示追加模式
    val fileWriter = new FileWriter("/stream_test/testLine.txt", true)
    
    try {
      dataList.foreach { line =>
        fileWriter.write(line + "\n")
      }
    } finally {
      fileWriter.close()
    }
  }
}