Spark Structured Streaming水印未正常工作：聚合重复记录问题排查

核心问题 #

你的水印未生效的根源是错误地在foreachBatch内部的静态DataFrame上使用withWatermark——流处理的水印必须定义在流式DataFrame的转换链中，才能让Spark维护跨微批的状态，实现延迟数据的合并。而你当前的写法中，withWatermark只作用于单个微批的静态数据，无法跟踪跨微批的状态，自然无法合并延迟数据。

另外，groupBy中同时使用timeslot和15分钟window可能存在时间范围不一致的问题，导致同一逻辑分组被拆分为多条记录。

修复步骤 #

1. 重构流处理链，将水印和聚合移到流层面

把数据解析、水印、聚合逻辑从foreachBatch内部移到主流处理链中，让Spark能维护跨微批的状态：

# 1. 从Kafka读取流数据
df_stream = self.spark.readStream.format('kafka') \
    .option("kafka.security.protocol", "SSL") \
    .option("kafka.ssl.truststore.location", self.ssl_truststore_location) \
    .option("kafka.ssl.truststore.password", self.ssl_truststore_password) \
    .option("kafka.ssl.keystore.location", self.ssl_keystore_location_bandwidth_intermediate) \
    .option("kafka.ssl.keystore.password", self.ssl_keystore_password_bandwidth_intermediate) \
    .option("kafka.bootstrap.servers", self.kafkaBrokers) \
    .option("subscribe", topic) \
    .option("startingOffsets", "latest") \
    .option("failOnDataLoss", "false") \
    .option("kafka.metadata.max.age.ms", "1000") \
    .option("kafka.ssl.keystore.type", "PKCS12") \
    .option("kafka.ssl.truststore.type", "PKCS12") \
    .load()

# 2. 解析JSON数据（需要提前定义好schema）
from pyspark.sql.functions import from_json, col, window, sum, to_json, struct, concat_ws
from pyspark.sql.types import StructType, StringType, LongType, TimestampType

# 假设原始value的JSON结构对应的schema
schema = StructType() \
    .add("applianceName", StringType()) \
    .add("timeslot", StringType()) \
    .add("sentOctets", StringType()) \
    .add("recvdOctets", StringType()) \
    .add("ts", TimestampType()) \
    .add("customer", StringType())

parsed_df = df_stream.selectExpr("CAST(value AS STRING)") \
    .select(from_json(col("value"), schema).alias("data")) \
    .select(
        col("data.applianceName"),
        col("data.timeslot"),
        col("data.sentOctets").cast(LongType()),
        col("data.recvdOctets").cast(LongType()),
        col("data.ts").alias("ts"),
        col("data.customer")
    )

# 3. 定义水印（关键：必须在流式DF上调用，且在groupBy之前）
watermarked_df = parsed_df.withWatermark("ts", "15 minutes")

# 4. 执行聚合
aggregated_df = watermarked_df.groupBy(
    "applianceName",
    "timeslot",
    "customer",
    window(col("ts"), "15 minutes")
).agg(
    sum("sentOctets").alias("sentOctets"),
    sum("recvdOctets").alias("recvdOctets")
).fillna(0)

# 5. 使用foreachBatch写入Kafka（此时只负责写入，不再做业务处理）
def write_to_kafka(ldf, batch_id):
    if not ldf.isEmpty():
        # 转换为Kafka所需的key-value格式
        kafka_output_df = ldf.select(
            concat_ws("|", col("applianceName"), col("timeslot"), col("customer")).alias("key"),
            to_json(struct(*ldf.columns)).alias("value")
        )
        kafka_output_df.write \
            .format("kafka") \
            .option("kafka.bootstrap.servers", self.kafkaBrokers) \
            # 补充Kafka SSL配置
            .option("kafka.security.protocol", "SSL") \
            .option("kafka.ssl.truststore.location", self.ssl_truststore_location) \
            .option("kafka.ssl.truststore.password", self.ssl_truststore_password) \
            .option("kafka.ssl.keystore.location", self.ssl_keystore_location_bandwidth_intermediate) \
            .option("kafka.ssl.keystore.password", self.ssl_keystore_password_bandwidth_intermediate) \
            .option("kafka.ssl.keystore.type", "PKCS12") \
            .option("kafka.ssl.truststore.type", "PKCS12") \
            .option("topic", output_topic) \
            .save()

# 启动流查询
query = aggregated_df.writeStream \
    .outputMode("update") \
    .trigger(processingTime='10 minutes') \
    .option("checkpointLocation", self.checkpoint) \
    .foreachBatch(write_to_kafka) \
    .start()

2. 对齐timeslot与window的时间范围

如果timeslot是自定义的时间分区字段，必须确保它的时间范围和window(ts, 15 minutes)完全一致（比如timeslot是15分钟的起始时间字符串）。否则建议直接去掉timeslot，用window的start或end字段作为时间分区键，避免分组冲突。

3. 移除不必要的count()调用

ldf.count()会触发全量数据扫描，严重影响流处理性能，空DataFrame写入Kafka不会产生任何数据，无需提前判断。

调试方法 #

• 查看Spark UI状态：在Streaming页面观察每个微批的状态大小，如果状态持续增长，说明水印未正确清理旧状态，检查水印字段是否为Timestamp类型、时间范围是否合理。
• 打印微批数据：在write_to_kafka中添加ldf.show()，查看每个微批的聚合结果，确认延迟数据是否被合并到已有窗口。
• 验证时间分组一致性：打印window的start和end字段，与timeslot对比，确认两者时间范围完全匹配。
• 模拟延迟数据：手动生成延迟10分钟以内的测试数据，观察是否被合并到对应的聚合记录，而非生成新记录。

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