Spark Streaming从Kafka导数据至PostgreSQL后数据丢失问题求助

你遇到的核心问题是使用了SaveMode.Overwrite模式写入数据库——这个模式会在每个Spark微批处理时，完全覆盖目标表的内容。也就是说，每次新的批数据过来，之前写入的所有数据都会被清空，只保留当前批的记录。这就是为什么你看到表数据只存在几秒（下一批处理就覆盖了），而停止应用后最后一批数据能留存的原因。

针对你的需求（流式写入PostgreSQL，基于Kafka键避免重复），我给你几个逐步的解决方案：

1. 切换为Append模式（基础版） #

如果你的Kafka消息不会重复（或者暂时不处理重复），直接把写入模式改成SaveMode.Append，这样每个微批的新数据会追加到表中，不会覆盖之前的内容：

censusRecordJavaDStream.map(e -> { 
    Row row = RowFactory.create(e.getAllValues()); 
    return row; 
}).foreachRDD(rdd -> { 
    Dataset<Row> censusDataSet = spark.createDataFrame(rdd, CensusRecord.getStructType()); 
    censusDataSet 
        .write() 
        .mode(SaveMode.Append) // 替换Overwrite为Append
        .jdbc("jdbc:postgresql:postgres", "census.census", connectionProperties); 
});

2. 实现Upsert逻辑（处理重复/主键冲突） #

因为你提到Kafka键是唯一ID，需要避免主键冲突或重复条目，这时候要用到PostgreSQL的ON CONFLICT语法实现Upsert（存在则更新，不存在则插入）。步骤如下：

前提准备 #

先确保你的PostgreSQL目标表census.census已经在唯一ID列（对应Kafka键的列）上创建了主键或唯一约束，比如：

ALTER TABLE census.census ADD PRIMARY KEY (your_unique_id_column);

修改代码实现Upsert #

Spark JDBC原生不支持直接Upsert，所以我们可以先把当前批的数据写入临时表，再通过SQL执行Upsert操作：

censusRecordJavaDStream.map(e -> { 
    Row row = RowFactory.create(e.getAllValues()); 
    return row; 
}).foreachRDD(rdd -> { 
    Dataset<Row> censusDataSet = spark.createDataFrame(rdd, CensusRecord.getStructType()); 
    
    // 1. 将当前批数据写入临时表（每次覆盖临时表内容）
    censusDataSet.write()
        .mode(SaveMode.Overwrite)
        .jdbc("jdbc:postgresql:postgres", "census.temp_census", connectionProperties);
    
    // 2. 执行Upsert：从临时表导入数据到目标表，冲突时更新字段
    String upsertSql = "INSERT INTO census.census SELECT * FROM census.temp_census " +
                       "ON CONFLICT (your_unique_id_column) DO UPDATE SET " +
                       "column1 = EXCLUDED.column1, " + // 替换成你需要更新的实际列
                       "column2 = EXCLUDED.column2, " +
                       "..."; // 列出所有需要更新的非主键列
    
    // 执行SQL
    spark.sql(upsertSql);
});

3. 优化批处理与偏移量配置 #

你提到完整数据集处理时表计数约1700，这确实和批处理大小相关，可以调整以下参数优化：

• 控制批处理数据量：设置spark.streaming.kafka.maxRatePerPartition，限制每个Kafka分区每个批拉取的最大记录数，避免批过大导致内存压力或写入延迟。
• 调整批处理间隔：通过spark.streaming.batchDuration设置批处理的时间间隔（比如10秒），平衡处理速度和资源占用。
• 手动管理Kafka偏移量：建议关闭自动提交偏移量（enable.auto.commit=false），在数据成功写入PostgreSQL后再手动提交偏移量，避免数据丢失或重复消费。可以通过KafkaUtils.createDirectStream的偏移量管理API实现。

额外注意事项 #

• 确保PostgreSQL的JDBC驱动版本和你的Spark、PostgreSQL版本兼容，避免出现奇怪的连接问题。
• 如果你的数据集很大，考虑开启Spark的动态分区写入，或者调整spark.sql.shuffle.partitions参数，优化写入性能。

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