带有partitionColumn的Spark JDBC Postgres会产生重复和缺失的行

Question

我正在使用具有 10 个连接的 Spark JDBC 从 POSTGRES 表中读取数据，并使用 RNO 作为分区列，边界范围从 0 到表中的行数。

该表是动态的，每秒都在增加。

第一步是计算行数，然后我使用这个值来设置 upperBound。

我们面临的问题是我们总是得到重复的行，或者缺少我希望看到但找不到的行。

val rows = calcCountRows(table)
val numPartitions = math.min(partitions, rows)

spark.read
  .format("jdbc")
  .options(database.connectionDetails())
  .option("fetchsize", 100)
  .option("dbtable",
    s"""
       |(select ROW_NUMBER() OVER(ORDER BY NULL) AS RNO, subQuery.* from $table subQuery) as "$table"
       |""".stripMargin)
  .option("partitionColumn", "RNO")
  .option("lowerBound", 0)
  .option("upperBound", rows)
  .option("numPartitions", numPartitions)
  .load()

当我从 DataFrame 获取数据时，有时会看到重复的条目，如果我尝试再次读取，我会看到不同的重复条目，甚至丢失条目。

我该如何处理这个问题？

Answer 1

通过 PostgreSQL 中的可重复读隔离增强读一致性

在 PostgreSQL 中处理并发事务时，确保数据一致性至关重要。可重复读隔离级别提供了一种可靠的方法来防止称为幻读的特定类型的不一致。

幻读：理解问题

当事务在单个事务内重新执行查询并且结果集意外不同时，就会发生幻读。当另一个事务同时插入与查询条件匹配的新行时，就会出现这种差异。由于 READ COMMITTED（默认隔离级别）仅保证事务启动后已提交数据的可见性，因此这些新插入的行可能在第一个查询中看不到，但出现在后续查询中，从而产生幻影效应。

Postgres MVCC 和快照隔离：确保读取稳定性

PostgreSQL 利用多版本并发控制 (MVCC) 来提供强读取一致性。 MVCC 在每个事务开始时创建数据库的只读快照。这些快照反映了数据库在特定时间点的状态，将事务与并发事务所做的更改隔离开来。

JDBC 连接和设置隔离级别

虽然您链接的 JDBC 文档可以作为有价值的参考，但以下是使用标准 JDBC 连接方法将隔离级别设置为 REPEATABLE_READ 的详细信息：

1. 使用驱动管理器：

Java

Connection conn = DriverManager.getConnection(
    "jdbc:postgresql://your-host:port/your-database",
    "username",
    "password");

conn.setTransactionIsolation(Connection.TRANSACTION_REPEATABLE_READ);

谨慎使用代码。

2. 使用连接池：

Java // 假设您已经使用 HikariCP 这样的库配置了连接池

DataSource dataSource = getDataSource();
Connection conn = dataSource.getConnection();

conn.setTransactionIsolation(Connection.TRANSACTION_REPEATABLE_READ);

谨慎使用代码。

主要考虑因素和优势

权衡：与 READ COMMITTED 相比，将隔离级别设置为 REPEATABLE_READ 可能会带来轻微的性能开销。然而，在避免幻读和维护数据库一致视图至关重要的情况下，增强数据一致性的好处通常会超过这种成本。 Postgres MVCC 实现：重要的是要记住，虽然 PostgreSQL 的 REPEATABLE_READ 可以防止幻读，但它仍然可能允许其他形式的序列化异常，具体取决于特定的事务模式和模式设计。始终彻底测试应用程序在并发工作负载下的行为，以确保数据完整性。