PySpark流重启后foreachBatch重复处理数据如何通过checkpoint避免

你只需要在writeStream配置中新增checkpointLocation参数，指定一个持久化的共享存储路径即可，该路径会持久化记录流任务的处理进度、偏移量信息，重启时会自动从上次结束的位置读取新数据，不会重复处理旧数据。

修改后的完整代码如下：

df = spark.readStream.option("readChangeFeed", "true").option("startingVersion", 2).load(tablePath)

def foreach_batch_function(df, epoch_id):
  print("epoch_id: ", epoch_id)
  df.write.mode("append").json("/mnt/sample/data/test/")

df.writeStream \
  .option("checkpointLocation", "/mnt/persistent/checkpoint/delta_cdf_stream/") \
  .foreachBatch(foreach_batch_function) \
  .start()

配置说明 #

• checkpointLocation指定的路径需要是所有Spark执行节点可访问的持久化存储路径，例如分布式文件系统HDFS、对象存储S3/ADLS、挂载的共享目录等，禁止使用单个节点的本地临时目录。
• 每个流任务需要对应唯一的checkpoint路径，多任务共用同一路径会导致进度记录冲突。
• 首次启动任务且checkpoint路径为空时，会按你配置的startingVersion=2开始读取数据；后续重启任务时会优先读取checkpoint中记录的已处理进度，自动从上次结束的位置继续读取新数据，不会重复处理历史数据。

额外注意事项 #

• 不要手动修改checkpoint路径下的文件，否则会破坏流任务的进度记录，导致数据重复或丢失。
• 如果需要重置任务从头读取数据，直接删除对应checkpoint路径的所有内容后重启任务即可。

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