Kafka Streams精确一次处理对接Spark Structured Streaming的隔离级别问题咨询

不需要移除Kafka Streams的Exactly once processing机制，以下是可行的替代方案：

1. 正确配置Spark Kafka消费者的隔离级别 #

Spark Kafka连接器支持通过带前缀的配置参数指定Kafka原生的隔离级别，你可能是因为配置方式错误导致看似无法设置。在读取Kafka流时，需将Kafka的isolation.level参数以kafka.为前缀添加到Spark的options中：

val streamDF = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "your-broker-list")
  .option("subscribe", "your-target-topic")
  .option("kafka.isolation.level", "read_committed") // 指定读取已提交的事务消息
  .option("startingOffsets", "latest") // 根据业务需求调整起始offset
  .load()

该配置适用于Spark 2.3及以上版本，确保Spark只消费Kafka Streams事务中已提交的消息，与Kafka Streams的Exactly once机制完全兼容。

2. 利用Delta Lake的ACID特性实现端到端精确一次写入 #

即使Spark无法通过隔离级别过滤未提交消息，Delta Lake本身的ACID事务和幂等性可以弥补这一点，保证最终写入S3的Delta表数据一致性：

• 使用结构化流的Exactly once写入模式：配置Spark流的写入模式为append，并开启Delta的事务支持：

streamDF
  .selectExpr("CAST(value AS STRING)")
  .writeStream
  .format("delta")
  .option("checkpointLocation", "s3://your-checkpoint-path") // 必须设置checkpoint保证容错
  .option("path", "s3://your-delta-table-path")
  .trigger(Trigger.ProcessingTime("1 minute"))
  .start()

• 自定义幂等写入逻辑：如果业务需要更精细的控制，可以通过foreachBatch结合Delta的merge操作，基于唯一业务键或Kafka offset实现幂等写入，避免重复数据：

def writeToDelta(batchDF: DataFrame, batchId: Long): Unit = {
  batchDF.createOrReplaceTempView("temp_data")
  spark.sql("""
    MERGE INTO delta.`s3://your-delta-table-path` t
    USING temp_data s
    ON t.unique_key = s.unique_key
    WHEN NOT MATCHED THEN INSERT *
  """)
}

streamDF
  .writeStream
  .foreachBatch(writeToDelta _)
  .option("checkpointLocation", "s3://your-checkpoint-path")
  .start()

3. 备选方案：Kafka Streams端输出事务确认信号 #

如果上述方案都无法满足需求，可以在Kafka Streams应用中，每当完成一个事务提交时，向一个单独的"transaction-commit" topic发送一条包含事务ID和offset范围的确认消息。Spark同时消费目标业务topic和这个确认topic，在流处理中关联两者，只处理已被确认提交的消息。

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