从一个非常大的表中获取数据

Question

我在 MySQL 数据库中有一个非常大的表，表 Users 中有 2 亿条记录。

我使用 JDBC 进行查询：

public List<Pair<Long, String>> getUsersAll() throws SQLException {
        Connection cnn = null;
        CallableStatement cs = null;
        ResultSet rs = null;
        final List<Pair<Long, String>> res = new ArrayList<>();
        try {
            cnn = dataSource.getConnection();
            cs = cnn.prepareCall("select UserPropertyKindId, login from TEST.users;");
            rs = cs.executeQuery();
            while (rs.next()) {
                res.add(new ImmutablePair<>(rs.getLong(1), rs.getString(2)));
            }
            return res;
        } catch (SQLException ex) {
            throw ex;
        } finally {
            DbUtils.closeQuietly(cnn, cs, rs);
        }
    }

接下来，我处理结果：

List<Pair<Long, String>> users= dao.getUsersAll();
            if (CollectionUtils.isNotEmpty(users)) {
                for (List<Pair<Long, String>> partition : Lists.partition(users, 2000)) {
                    InconsistsUsers.InconsistsUsersCallable callable = new InconsistsUsers.InconsistsUsersCallable (new ArrayList<>(partition));
                    processExecutor.submit(callable);
                }
            }

但是由于表非常大并且全部卸载到内存中，我的应用程序崩溃并出现错误：

com.mysql.jdbc.exceptions.jdbc4.CommunicationsException：通信链路故障

从服务器成功接收到的最后一个数据包是 105,619 毫秒前。

如何分批接收数据并按优先级顺序处理，不至于一次将所有结果上传到内存中？可以创建游标并将数据上传到非阻塞队列并在数据到达时对其进行处理。如何做到这一点？

更新：

我的数据库结构：https://www.db-fiddle.com/f/v377ZHkG1YZcdQsETtPm9L/3

当前算法：

1. 从Users表中获取所有数据用户：select UserPropertyKindId, login from Users;

2. 这个结果被分成2000对并提交给ThreadPoolTaskExecutor：

   List<Pair<Long, String>> users= dao.getUsersAll();
   
   if (CollectionUtils.isNotEmpty(users)) {
       for (List<Pair<Long, String>> partition : Lists.partition(users, 2000)) {
           InconsistsUsers.InconsistsUsersCallable callable = new InconsistsUsers.InconsistsUsersCallable(new ArrayList<>(partition));
           processExecutor.submit(callable));
       }
   }
   

3. 在 callable 中为每一对做两个查询：

   第一个查询：

   select distinct entityId 
   from UserPropertyValue 
   where userPropertyKindId= ? and value = ? -- value its login from Users table
   

   第二次查询：

   select UserIds 
   from UserPropertyIndex 
   where UserPropertyKindId = ? and Value = ?
   

可能有两种情况：

1. 第一个查询结果为空：记录，发送通知，继续下一个对
2. 第二次查询的结果不等于第一次查询的结果（varbinary 数据已解码。存储了编码的 entityId）。然后记录，发送通知，转到下一对。

我不能改变基地的结构。我必须在 Java 代码方面进行的所有操作。

Answer 1

您应该将 mysql 结果集限制为每个请求 2000 个，而不是 Java 端的 Lists.partition(users, 2000)。

select UserPropertyKindId, login from TEST.users limit <offset>, 2000;

更新：正如 Raymond Nijland 在下面的评论中提到的，如果偏移量太大，查询可能会显着减慢。

一种解决方法可能是不使用偏移量，而是引入 where 语句，例如 where id > last_user_id。

由于@All_safe 在下面评论，不存在自动增量ID，大限制偏移的另一种解决方法是：仅在子查询中获取主键，然后连接回主表。这将迫使mysql不做早期行查找，这是大偏移限制的主要问题。

但您的原始查询仅获取主键列，我认为早期行查找不适用。

Answer 2

你应该在几个层面上处理这个问题：

JDBC 驱动程序获取大小

JDBC 有一个Statement.setFetchSize() 方法，它指示在您从 JDBC 获取数据之前，JDBC 驱动程序将预取多少行。请注意，MySQL JDBC 驱动程序并没有真正正确地实现这一点，但您可以设置 setFetchSize(Integer.MIN_VALUE) 以防止它一次性获取所有行。 See also this answer here.

注意，您也可以使用 useCursorFetch 激活连接上的功能

你自己的逻辑

您不应将整个用户列表放入内存中。您现在正在做的是从 JDBC 收集所有行，然后稍后使用 Lists.partition(users, 2000) 对您的列表进行分区。这是朝着正确的方向发展，但你还没有做对。相反，这样做：

try (ResultSet rs = cs.executeQuery()) {
    while (rs.next()) {
        res.add(new ImmutablePair<>(rs.getLong(1), rs.getString(2)));
    }

    // Process a batch of rows:
    if (res.size() >= 2000) {
        process(res);
        res.clear();
    }
}

// Process the remaining rows
process(res);

这里的重要信息是不要将所有行加载到内存中然后批量处理它们，而是在从 JDBC 流式传输行时直接处理它们。

Answer 3

我也遇到过类似的情况。我正在从 MySQL 数据库中读取数据并将其复制到 MS SQL Server 数据库中。不是2亿，每天只有400万。但是我收到了与通信链接故障相同的错误消息。我可以通过设置 PreparedStatement.setFetchSize(Integer.MIN_VALUE); 的 fetchsize 来解决它 于是通讯链路故障就消失了。我知道，这并不能解决您的列表问题。

Answer 4

您可以将您的优先级纳入查询中 例如，WHERE my_priority = 1 ORDER BY my_sub_priority DESC

就像 Jacob 所说，使用限制 LIMIT 0, 2000

您可能可以分解consistent_users 中的逻辑以查找特定缺陷，然后根据EXPLAIN 中获得的见解优化这些查询。也许 find_user_defect(defect) 类型的方法将帮助您按组处理用户。