Spark Structured Streaming用AvailableNow Trigger读写Kafka到S3异常问题

现象解释：S3仅在任务完成后生成对象 #

这是Delta Lake的原子提交机制导致的正常行为：

在AvailableNow触发器模式下，Spark会将整个周期内的所有微批处理视为一个全局事务。每个微批的处理结果会先写入S3的临时目录，直到所有微批处理完成、偏移量确认无误后，才会一次性将所有临时文件移动到正式路径，并更新Delta表的元数据。这种设计是为了保证数据一致性，避免出现中间状态的脏数据。

前序微批数据丢失的原因与修复 #

1. coalesce(2)误用导致数据覆盖 #

coalesce是窄依赖操作，仅合并现有分区而不重新洗牌数据。在多微批场景下，强制将每个微批的数据合并为2个分区，会导致不同微批的分区数据写入同一文件路径，最终只有最后一批数据被保留。

修复方案：
将coalesce(2)替换为repartition(2)。repartition是宽依赖操作，会为每个微批重新分区数据，确保各微批输出文件独立，不会相互覆盖：

# 替换原 coalesce 代码
df = df.repartition(2)

2. AvailableNow事务边界与偏移量管理 #

AvailableNow会将所有微批的偏移量作为单个事务提交到checkpoint。如果转换逻辑包含状态操作（如聚合、关联），或startingOffsets配置错误，可能导致前序微批数据未被正确写入。

检查要点：

• 确认startingOffsets配置：若需消费所有未处理数据，应设置为earliest或依赖checkpoint自动恢复偏移量（推荐）；若设置为latest，每次启动只会从最新偏移量开始消费。
• 确保checkpoint路径具备正确读写权限，Spark能正常记录偏移量与事务状态。

3. 数据完整性验证方法 #

• 查看Spark UI的Streaming标签，核对每个微批的输入记录数与处理记录数，确认前序微批数据已被处理。
• 任务完成后，执行以下代码统计Delta表总记录数，与Kafka消费偏移量对应的记录数对比：

total_records = spark.read.format("delta").load(cf.sink_location).count()
print(f"Total records in Delta table: {total_records}")

每3小时运行任务的实现 #

结合调度工具（如Airflow、Oozie）定时启动任务即可。每次启动时，Spark会从checkpoint记录的偏移量开始，处理所有可用Kafka数据，完成后自动停止，无需额外修改代码逻辑。

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