PySpark流处理写入Delta表无数据问题排查：Kafka读流处理后无输出

我帮你梳理下可能导致这个问题的几个关键点，结合你的代码和场景来看：

1. 窗口聚合+水印的延迟输出特性（最可能的原因） #

你用了2分钟窗口聚合+1分钟水印延迟，而且输出模式是append——这是核心问题！在append模式下，Spark不会立即把窗口聚合的结果写入Delta表，必须等到水印时间推进到窗口结束时间之后，才会认为这个窗口的所有数据都已经到齐，才会将该窗口的聚合结果持久化到输出表中。

举个例子：假设某个窗口的时间范围是00:00-00:02，水印延迟是1分钟，那么只有当水印时间超过00:03（窗口结束时间+水印延迟）时，这个窗口的query_count结果才会被写入Delta表。你在display(df)时看到的是Spark内存中实时维护的聚合状态，但这些状态还没满足输出条件，所以不会落地到Delta表。

验证方法： #

临时把输出模式改成update，然后重新运行任务。如果update模式下能看到数据写入Delta表，那就能确认是append模式的延迟机制导致的，不是代码逻辑问题。

2. start()方法的路径参数错误 #

你的代码里调用了query.start(OUTPUT_PATH)，但PySpark的DataStreamWriter.start()方法不需要传入路径参数——正确的做法是把输出路径配置在writeStream的链中。

修改建议： #

在writeStream部分添加.path(OUTPUT_TABLE_PATH)（注意你的常量是OUTPUT_TABLE_PATH，避免和OUTPUT_PATH混淆），然后调用start()时不带参数：

# 在run函数的writeStream部分添加path配置
return (spark
        ... # 前面的处理逻辑
        .writeStream
        .outputMode(OUTPUT_MODE)
        .path(OUTPUT_TABLE_PATH)  # 新增这一行，指定输出路径
        # .partitionBy("date", "hour")
        .format(OUTPUT_FORMAT)
        .option("mergeSchema", "true")
        .option("checkpointLocation", CHECKPOINT_LOCATION))

# 调用start时不需要传参数
query = run(spark, "2 minutes", "1 minuteS")
query.start().awaitTermination()

3. 权限与路径配置问题 #

不管是在Kubernetes还是Databricks环境，要确保Spark对以下路径有读写权限：

• CHECKPOINT_LOCATION（wasbs路径）：检查点需要写入状态信息，没有权限会导致任务异常
• OUTPUT_TABLE_PATH：Delta表的存储路径，没有写权限的话，即使有输出也无法落地

如果是Azure Blob Storage，还要确认Spark配置了正确的存储账户密钥或SAS Token，否则会出现有权限但无法写入的情况。

4. 检查点状态残留 #

如果之前运行过失败的任务，检查点路径中残留的旧状态可能会干扰新任务的运行。可以尝试删除检查点路径，然后重新启动任务，看是否能正常写入数据。

5. 事件时间字段的正确性 #

确认输入数据的timestamp字段是事件时间（即数据产生的时间），而不是Spark处理数据的时间。如果timestamp字段格式不正确或不是事件时间，会导致窗口和水印计算异常，进而无法触发输出。你可以在display(df)时检查period_start、date等字段是否符合预期，来验证这一点。

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