Spark Streaming写入输出全量数据而非增量数据问题求助

• 核心原因1：流任务状态未持久化保留：每次运行流任务时如果是独立启动新的Spark应用，且没有稳定复用同一个checkpointLocation路径，流任务无法记录已经处理过的源文件，每次启动都会扫描路径下所有文件作为全量数据处理。
• 核心原因2：时间戳变量作用域错误：代码中的currentTimestamp是在Driver端定义流查询逻辑时就生成的固定值，而非每次foreachBatch触发时动态生成的新时间戳，单次调度Trigger.Once任务时，时间戳生成逻辑位置错误会导致写入逻辑异常。
• 额外可能原因：源文件的修改时间频繁更新，导致文件源将旧文件识别为新增未处理文件，触发重复读取。

1. 固定复用checkpoint路径：确保每次调度运行流任务时，checkpointLocation的路径完全一致，不要每次更换路径或清理路径下的文件，流任务会在该路径下存储已处理文件的元数据，仅读取自上次处理完成后新增的文件。
2. 修正时间戳的生成逻辑：将currentTimestamp的生成放到foreachBatch内部，确保每次批次处理时都生成对应执行时间的新路径，修正后的代码如下：

import org.apache.spark.sql.streaming.Trigger
import org.apache.spark.sql.SaveMode

spark.readStream
  .schema(spark.read.load("s3a://bucketName/inPath").schema)
  .parquet("s3a://bucketName/inPath")
  .writeStream
  .trigger(Trigger.Once)
  // 注意原路径拼写错误，已将bucekt修正为bucket
  .option("checkpointLocation", "s3a://bucketName/outPath/checkpoint")
  .foreachBatch { (batchDF: DataFrame, batchId: Long) =>
      // 时间戳移到foreachBatch内部动态生成
      val currentTimestamp = System.currentTimeMillis()
      batchDF
      .write
      .mode(SaveMode.Append)
      .parquet(s"s3a://bucketName/outPath/${currentTimestamp}")
  }.start()
  .awaitTermination()

3. 可选配置优化：如果需要避免读取修改时间更新的旧文件，可以添加文件源参数限制读取的文件修改时间范围，示例：

spark.readStream
  .option("maxFileAge", "7d") // 仅读取最近7天内修改的文件
  .schema(spark.read.load("s3a://bucketName/inPath").schema)
  .parquet("s3a://bucketName/inPath")

4. 逻辑验证：首次运行任务会全量读取现有文件写入第一个时间戳文件夹，后续向源路径新增文件后再运行任务，只会将新增文件写入新的时间戳文件夹，不会重复读取旧文件。

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