Spark Structured Streaming重启后dropDuplicates失效问题求助

问题本质 #

你碰到的核心是流处理重启后的端到端Exactly-Once语义缺失：Spark的Checkpoint只记录自身的处理状态，但没和NATS Jetstream的消费位移做强绑定；而dropDuplicates是基于Spark内存/Checkpoint的状态去重，一旦应用重启，未完全提交的微批状态丢失，自然识别不了历史处理过的消息。Watermark是用来处理迟到数据的，和跨重启去重完全不相关，还会额外增加状态管理开销导致变慢。

下面是几个落地性强的解决方案：

方案1：Delta Lake 幂等写入（最推荐） #

利用Delta Lake的ACID特性，用merge操作替代直接writeStream写入，基于消息的唯一标识（比如NATS消息的Message ID，或者你用的dateTime+content组合键）做upsert，重复消息会自动被过滤。

Java代码示例 #

// 假设你的数据流是Dataset<Row> streamData，包含msgId、dateTime、content等字段
streamData.writeStream()
    .foreachBatch((batchDF, batchId) -> {
        // 执行merge操作：如果msgId已存在则跳过，否则插入
        DeltaTable.forPath(spark, "/path/to/delta-table")
            .as("target")
            .merge(
                batchDF.as("source"),
                "target.msgId = source.msgId" // 这里用唯一键匹配
            )
            .whenNotMatched()
            .insertAll()
            .execute();
    })
    .option("checkpointLocation", "/path/to/checkpoint")
    .start()
    .awaitTermination();

优势 #

• 无需额外依赖，直接利用Delta本身特性
• 即使NATS重推消息，Delta会自动过滤重复，保证最终一致性
• 性能开销远低于watermark+dropDuplicates

方案2：绑定NATS ACK与Spark Checkpoint提交时机 #

问题根源之一是NATS在Spark微批未完全提交Checkpoint前就收到了ACK，或者Spark挂了导致NATS没收到ACK。可以自定义消费逻辑，等Delta写入完成且Checkpoint提交后，再批量ACK NATS消息。

实现思路 #

1. 不用Spark的NATS Source，直接用NATS Java客户端手动消费Jetstream消息，将消息拉取到Spark的Dataset
2. 在foreachBatch中完成Delta写入后，批量ACK这批消息
3. 将消费位移（比如Jetstream的sequence number）写入Checkpoint，重启时从Checkpoint读取位移继续消费

关键代码片段 #

// 初始化NATS Jetstream客户端
JetStream js = nc.jetStream();
Consumer consumer = js.subscribe("subject", ConsumerOpts.builder()
    .durable("durable-name") // 持久化消费者，保证位移不丢失
    .build());

// 从Checkpoint读取上次消费的最后sequence
long lastSeq = readLastSequenceFromCheckpoint("/path/to/checkpoint");

// 拉取消息（从lastSeq之后开始）
List<Message> messages = new ArrayList<>();
Message msg;
while ((msg = consumer.nextMessage(Duration.ofSeconds(10))) != null) {
    if (msg.metaData().streamSequence() > lastSeq) {
        messages.add(msg);
        lastSeq = msg.metaData().streamSequence();
    }
}

// 转换为Dataset处理
Dataset<Row> batchDF = spark.createDataFrame(messages, Message.class).select(...);

// 写入Delta
batchDF.write().format("delta").mode("append").load("/path/to/delta-table");

// 批量ACK消息
for (Message m : messages) {
    m.ack();
}

// 更新Checkpoint中的lastSeq
writeLastSequenceToCheckpoint("/path/to/checkpoint", lastSeq);

优势 #

• 实现严格的Exactly-Once语义，从消费到写入完全绑定
• 避免NATS无意义的重推

方案3：外部状态存储记录已处理消息 #

如果消息没有全局唯一ID，或者需要跨多个应用去重，可以用Redis/另一个Delta表作为状态存储，记录已处理消息的标识（比如dateTime+content的哈希值），每次处理前先过滤。

Java代码示例（用Delta表做状态存储） #

// 定义状态表路径
String stateTablePath = "/path/to/processed-messages";

// 读取数据流
Dataset<Row> streamData = spark.readStream().format("nats").option(...).load();

// 生成消息唯一标识哈希
Dataset<Row> dataWithHash = streamData.withColumn(
    "msg_hash",
    sha2(concat(col("dateTime"), col("content")), 256)
);

// 处理并写入主Delta表
dataWithHash.writeStream()
    .foreachBatch((batchDF, batchId) -> {
        // 读取已处理的哈希表
        Dataset<Row> processedHashes = spark.read().format("delta").load(stateTablePath);
        
        // 过滤已处理的消息
        Dataset<Row> newData = batchDF.join(
            processedHashes,
            batchDF.col("msg_hash").equalTo(processedHashes.col("msg_hash")),
            "left_anti" // 只保留未处理的消息
        );
        
        // 写入主表
        newData.drop("msg_hash").write().format("delta").mode("append").load("/path/to/main-table");
        
        // 将本次处理的哈希写入状态表
        newData.select("msg_hash").write().format("delta").mode("append").load(stateTablePath);
    })
    .option("checkpointLocation", "/path/to/checkpoint")
    .start()
    .awaitTermination();

注意事项 #

• 定期清理状态表的旧数据（比如用Delta的vacuum），避免存储膨胀
• 可以用Redis代替Delta表，查询性能更高，适合高吞吐量场景

为什么之前的方案没用？ #

• dropDuplicates：依赖Spark的状态存储，这个状态是和Checkpoint绑定的，但如果应用在微批中途挂了，Checkpoint中的状态可能未完全持久化，重启后丢失了已处理消息的记录
• Watermark：是用来清理迟到数据的状态，只能处理窗口内的重复，无法跨重启保留全局去重状态，还会增加状态管理的开销导致应用变慢

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