Spark Structured Streaming中如何用batchId实现Exactly-Once语义？

Spark Structured Streaming的foreachBatch默认仅提供至少一次（at-least-once）写入保证，当流作业重启或微批重试时，可能会导致同一份数据重复写入到数据库。要实现恰好一次（exactly-once），核心是借助传入foreachBatch函数的epoch_id（即batchId）做幂等写入——因为每个微批的epoch_id是唯一且严格递增的，我们可以基于它确保同批次数据只会被写入一次。

具体实现步骤 #

1. 在目标数据库创建批次记录表

首先需要在你的SQL数据库中创建一张专门记录已处理批次的表，用来跟踪哪些epoch_id已经成功完成写入：

CREATE TABLE processed_batches (
    batch_id BIGINT PRIMARY KEY,
    processed_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

batch_id设为主键，确保同一个批次ID不会被重复记录。

2. 修改JDBC写入函数，加入幂等校验和原子操作

修改write_to_db_using_jdbc函数，核心逻辑是：先校验当前批次是否已处理，未处理则在同一事务中完成业务数据写入和批次记录，保证原子性（要么都成功，要么都失败）。

示例代码如下：

def write_to_db_using_jdbc(df, epoch_id):
    # 数据库连接配置
    db_url = "jdbc:mysql://your-db-host:3306/your-db"
    db_properties = {
        "user": "your-user",
        "password": "your-password",
        "driver": "com.mysql.cj.jdbc.Driver"
    }
    
    # 1. 检查当前批次是否已处理
    check_query = f"SELECT COUNT(*) FROM processed_batches WHERE batch_id = {epoch_id}"
    check_df = spark.read.jdbc(url=db_url, table=f"({check_query}) AS temp", properties=db_properties)
    processed_count = check_df.collect()[0][0]
    
    if processed_count > 0:
        print(f"批次 {epoch_id} 已处理，跳过写入")
        return
    
    # 2. 开启数据库事务，原子执行写入和批次记录
    connection = None
    try:
        # 获取JDBC连接
        connection = db_properties["driver"].newInstance().connect(db_url, db_properties)
        connection.setAutoCommit(False)
        
        # 写入业务数据到目标表（假设目标表名为event_counts）
        df.write.jdbc(url=db_url, table="event_counts", mode="append", properties=db_properties, connection=connection)
        
        # 记录当前批次为已处理
        insert_batch_sql = f"INSERT INTO processed_batches (batch_id) VALUES ({epoch_id})"
        statement = connection.createStatement()
        statement.executeUpdate(insert_batch_sql)
        
        # 提交事务
        connection.commit()
        print(f"批次 {epoch_id} 写入成功")
    except Exception as e:
        # 出错则回滚事务
        if connection:
            connection.rollback()
        print(f"批次 {epoch_id} 写入失败，已回滚: {str(e)}")
        raise e
    finally:
        # 关闭连接
        if connection:
            connection.close()

3. 额外注意事项

• 事务原子性：必须保证业务数据写入和批次ID记录在同一个事务中，否则可能出现数据写入但批次未记录的情况，导致后续重试时重复写入。
• 检查点配合：确保你的流作业配置了checkpointLocation，Spark会在检查点中记录批次进度，结合批次记录表的校验，可以双重避免重复处理。
• 分布式写入适配：如果DataFrame是分布式的，默认df.write.jdbc是分区并行写入，可先将数据写入临时表，再通过事务合并到目标表并记录批次ID，避免分布式场景下的事务问题。
• 业务表唯一约束：如果业务场景允许，也可以在业务表中加入batch_id字段，设置业务字段+batch_id为联合唯一约束，利用数据库约束自动过滤重复数据，省去单独的批次记录表。

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