PySpark流写入器Input Vs process rows显示为0 无法写入Delta表

以下是可直接落地的排查步骤：

1. 流数据源校验 #

• 确认readStream配置正确性，以Kafka数据源为例，检查bootstrap服务地址、topic名称、消费组ID、偏移量配置是否匹配实际业务环境。若配置了startingOffsets="latest"，流启动后无新消息进入时不会读取历史数据，直接出现处理行数为0的情况。
• 临时替换输出端测试数据源连通性：将原有写入逻辑替换为控制台输出，确认数据源本身有数据返回：

df.writeStream.format("console").start().awaitTermination()

2. 写入逻辑校验 #

• 确认目标Delta表的路径/表名配置正确，且Spark作业的执行身份对目标存储路径有读写权限。
• 检查是否存在全局过滤逻辑把所有数据过滤：可通过foreachBatch打印每批次数据量确认：

df.writeStream.foreachBatch(lambda batch_df, batch_id: print(f"批次{batch_id}数据量：{batch_df.count()}")) \
  .start().awaitTermination()

• 确认触发器配置符合预期：若配置了trigger(once=True)/trigger(availableNow=True)，运行时无新数据会直接结束作业无写入；连续触发模式下需确认作业未被意外终止。

3. Schema兼容校验 #

• Delta表默认开启Schema校验，若流DataFrame的Schema与目标表不兼容，部分低版本Delta会静默丢弃所有数据。可临时开启Schema合并测试是否能正常写入：

df.writeStream.format("delta") \
  .option("mergeSchema", "true") \
  .option("checkpointLocation", "你的检查点路径") \
  .start("目标Delta表路径")

• 确认字段名、数据类型、大小写（大小写敏感的Spark集群环境下）完全匹配目标表结构。

4. 检查点配置校验 #

• 若复用了历史作业的检查点路径，之前已提交的偏移量可能已到最新位点，导致新作业无新数据可消费，可更换全新的检查点路径，搭配startingOffsets="earliest"测试消费逻辑。
• 确认检查点路径无损坏的元数据文件，作业执行身份对该路径有读写权限。

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