如何实现Spark Streaming接续Spark Batch已处理进度运行

直接把批处理生成的checkpoint复用到流任务上，完全实现不了接续处理的效果，别踩这个坑。
不管是老版本DStream API的Spark Streaming，还是现在主流的Structured Streaming，批计算和流计算的checkpoint从存储结构到元数据语义完全不兼容：批任务的checkpoint本质是存shuffle中间态、任务容错断点的临时快照，根本不会记录「哪些文件已经被消费过」这类消费进度信息，流任务读到这种不兼容的checkpoint要么直接启动失败，要么直接全目录递归扫描从头消费，根本跳不过批处理已经跑过的存量文件。

• 批处理的checkpoint设计目标是给当前运行的批任务做失败重试用的，存的都是RDD/DataFrame的计算中间态、分区调度信息，任务跑完之后这部分数据很多场景下会被自动清理，从设计上就不支持跨任务、跨计算模式传递消费进度。
• Structured Streaming对接parquet文件源的时候，会在checkpoint目录下的sources/0/路径维护一套专属的offset日志，记录所有已经被消费的文件路径、文件修改时间、对应批次的偏移范围，这套日志格式是流任务独有的，批处理运行时根本不会生成这部分内容，自然也谈不上复用。
• 你的存储结构是单月目录10TB量级，单目录下文件数通常能到几十万甚至百万级，如果硬塞不兼容的checkpoint启动流任务，第一次启动光递归列目录的耗时就可能达到小时级，完全达不到预期效果。

不用自己维护全量已处理文件列表——10TB量级的文件全量列表光存就要占几百MB，每次流触发批次都要做全量路径比对，性能会差到根本跑不动，业内基本用两种方案搞定批流接续：

方案1：最小改动低成本方案（90%以上场景选这个） #

如果你跑批的目的只是为了快速跑完存量数据，不想等流任务慢慢按批次拉存量（毕竟10TB数据批处理可能20分钟跑完，流任务按单批次限流跑可能要1小时），操作逻辑非常简单：

1. 正常跑批处理存量数据，跑完之后记录批处理任务扫描到的最大文件修改时间戳就行，不用记全量文件列表。
2. 启动流任务的时候，直接加一个参数option("modifiedAfter", 你记录的最大时间戳)，流任务第一次启动就只会扫描修改时间晚于这个时间戳的新增文件，自动跳过所有批处理已经跑过的文件。
3. 第一次启动成功后，流任务会自动把这个起始消费位置写入自己的checkpoint，后续任务重启不需要再带这个参数，会自动接续消费进度，完全不用人工干预。

这个方案和你的存储规范完全适配：你的文件是原子写入（先写临时文件再rename到目标目录），不会出现文件写了一半被扫到、或者文件修改时间乱跳的问题，时间戳边界的精度完全能满足要求，不会漏读也不会重复读。
如果你连单独跑批的步骤都想省，直接启动流任务就行，把latestFirst设为false，根据集群处理能力调大maxFilesPerTrigger参数（比如单批次能扛住2000个文件就设2000），流任务会自动从最旧的文件开始消费，先把存量全跑完再无缝接每分钟的新增数据，全程checkpoint自动维护进度，连时间戳都不用自己记。

方案2：高一致性要求场景方案 #

如果你的场景是金融级精度要求，完全不能接受时间戳边界可能带来的极端误差（比如不同节点时钟不同步导致个别文件时间戳偏早/偏晚），可以手动给流任务灌初始offset：

1. 批处理跑完之后，拿到所有已处理文件的路径列表。
2. 按照你当前用的Spark版本对应的文件源offset序列化格式，把这些文件路径手动写入流任务checkpoint目录下的初始offset日志文件里，相当于提前告诉流任务这些文件已经处理过了。
3. 正常启动流任务即可，后续流任务会自己维护offset日志，不需要每次批次都做文件过滤。
   这个方案操作成本稍高，需要对齐对应Spark版本的内部序列化逻辑，非强一致要求的场景没必要用。

• 不要跨Spark大版本复用checkpoint，就算是流任务自己生成的checkpoint，跨大版本（比如3.2升到3.3）都可能出现序列化不兼容的问题，更别说批任务生成的checkpoint。
• 流任务读取路径直接指到月度目录的上层根路径就行，文件源会自动感知新建的月度目录，不需要你每次新月度手动加路径配置。
• 保持cleanSource参数为默认的off就好，不要开自动删除/归档已处理文件的逻辑，符合你文件写入后不可移动的要求。

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