发布于 2024-07-08 23:15:58

10 亿数据如何最快速插入 MySQL

原文地址 www.51cto.com

要首先确认约束条件，才能设计方案。确定面试官主要想问的方向，例如 1T 文件如何切割为小文件，虽是难点，然而可能不是面试官想考察的问题。

最快的速度把 10 亿条数据导入到数据库，首先需要明确一下，10 亿条数据什么形式存在哪里，每条数据多大，是否有序导入，是否不能重复，数据库是否是 MySQL？

假设有如下约束

10 亿条数据，每条数据 1 Kb
数据内容是非结构化的用户访问日志，需要解析后写入到数据库
数据存放在 Hdfs 或 S3 分布式文件存储里
10 亿条数据并不是 1 个大文件，而是被切分为多个文件，后缀标记顺序
要求有序导入，尽量不重复
数据库是 MySQL

首先考虑 10 亿数据写到 MySQL 单表可行吗？

数据库单表能支持 10 亿吗？

答案是不能，单表推荐的值是 2000W 以下。这个值怎么计算出来的呢？

MySQL 索引数据结构是 B + 树，全量数据存储在主键索引，也就是聚簇索引的叶子结点上。B + 树插入和查询的性能和 B + 树层数直接相关，2000W 以下是 3 层索引，而 2000w 以上则可能为四层索引。

Mysql b + 索引的叶子节点每页大小 16K。当前每条数据正好 1K，所以简单理解为每个叶子节点存储 16 条数据。b + 索引每个非叶子节点大小也是 16K，但是其只需要存储主键和指向叶子节点的指针，我们假设主键的类型是 BigInt，长度为 8 字节，而指针大小在 InnoDB 中设置为 6 字节，这样一共 14 字节，这样一个非叶子节点可以存储 16 * 1024/14=1170。

也就是每个非叶子节点可关联 1170 个叶子节点，每个叶子节点存储 16 条数据。由此可得到 B + 树索引层数和存储数量的表格。2KW 以上索引层数为 4 层，性能更差。

图片

为了便于计算，我们可以设计单表容量在 1KW，10 亿条数据共 100 个表。

如何高效的写入数据库

单条写入数据库性能比较差，可以考虑批量写入数据库，批量数值动态可调整。每条 1K，默认可先调整为 100 条批量写入。

批量数据如何保证数据同时写成功？MySQL Innodb 存储引擎保证批量写入事务同时成功或失败。

写库时要支持重试，写库失败重试写入，如果重试 N 次后依然失败，可考虑单条写入 100 条到数据库，失败数据打印记录，丢弃即可。

此外写入时按照主键 id 顺序顺序写入可以达到最快的性能，而非主键索引的插入则不一定是顺序的，频繁地索引结构调整会导致插入性能下降。最好不创建非主键索引，或者在表创建完成后再创建索引，以保证最快的插入性能。

是否需要并发写同一个表

不能

并发写同一个表无法保证数据写入时是有序的。
提高批量插入的阈值，在一定程度上增加了插入并发度。无需再并发写入单表

MySQL 存储引擎的选择

Myisam 比 innodb 有更好的插入性能，但失去了事务支持，批量插入时无法保证同时成功或失败，所以当批量插入超时或失败时，如果重试，势必对导致一些重复数据的发生。但是为了保证更快的导入速度，可以把 myisam 存储引擎列为计划之一。

图片

从数据可以看到批量写入明显优于单条写入。并且在 innodb 关闭即时刷新磁盘策略后，innodb 插入性能没有比 myisam 差太多。

innodb_flush_log_at_trx_commit: 控制 MySQL 刷新数据到磁盘的策略。

默认 = 1，即每次事务提交都会刷新数据到磁盘，安全性最高不会丢失数据。
当配置为 0、2 会每隔 1s 刷新数据到磁盘，在系统宕机、mysql crash 时可能丢失 1s 的数据。

考虑到 Innodb 在关闭即刷新磁盘策略时，批量性能也不错，所以暂定先使用 innodb（如果公司 MySQL 集群不允许改变这个策略值，可能要使用 MyIsam 了。）。线上环境测试时可以重点对比两者的插入性能。

要不要进行分库

mysql 单库的并发写入是有性能瓶颈的，一般情况 5K TPS 写入就很高了。

当前数据都采用 SSD 存储，性能应该更好一些。但如果是 HDD 的话，虽然顺序读写会有非常高的表现，但 HDD 无法应对并发写入，例如每个库 10 张表，假设 10 张表在并发写入，每张表虽然是顺序写入，由于多个表的存储位置不同，HDD 只有 1 个磁头，不支持并发写，只能重新寻道，耗时将大大增加，失去顺序读写的高性能。

所以对于 HDD 而言，单库并发写多个表并不是好的方案。回到 SSD 的场景，不同 SSD 厂商的写入能力不同，对于并发写入的能力也不同，有的支持 500M/s，有的支持 1G/s 读写，有的支持 8 个并发，有的支持 4 个并发。在线上实验之前，我们并不知道实际的性能表现如何。

所以在设计上要更加灵活，需要支持以下能力

支持配置数据库的数量
支持配置并发写表的数量，（如果 MySQL 是 HDD 磁盘，只让一张表顺序写入，其他任务等待）

通过以上配置，灵活调整线上数据库的数量，以及写表并发度，无论是 HDD 还是 SSD，我们系统都能支持。不论是什么厂商型号的 SSD，性能表现如何，都可调整配置，不断获得更高的性能。这也是后面设计的思路，不固定某一个阈值数量，都要动态可调整。

接下来聊一下文件读取，10 亿条数据，每条 1K，一共是 931G。近 1T 大文件，一般不会生成如此大的文件。所以我们默认文件已经被大致切分为 100 个文件。每个文件数量大致相同即可。为什么切割为 100 个呢？切分为 1000 个，增大读取并发，不是可以更快导入数据库吗？刚才提到数据库的读写性能受限于磁盘，但任何磁盘相比写操作，读操作都要更快。尤其是读取时只需要从文件读取，但写入时 MySQL 要执行建立索引，解析 SQL、事务等等复杂的流程。所以写的并发度最大是 100，读文件的并发度无需超过 100。

更重要的是读文件并发度等于分表数量，有利于简化模型设计。即 100 个读取任务，100 个写入任务，对应 100 张表。

如何保证写入数据库有序

既然文件被切分为 100 个 10G 的小文件，可以按照文件后缀 + 在文件行号作为记录的唯一键，同时保证同一个文件的内容被写入同一个表。例如

index_90.txt 被写入数据库 database_9，table_0 ，
index_67.txt 被写入数据库 database_6，table_7。

这样每个表都是有序的。整体有序通过数据库后缀 + 表名后缀实现。

如何更快地读取文件

10G 的文件显然不能一次性读取到内存中，场景的文件读取包括

Files.readAllBytes 一次性加载内内存
FileReader+ BufferedReader 逐行读取
File+ BufferedReader
Scanner 逐行读取
Java NIO FileChannel 缓冲区方式读取

在 MAC 上，使用这几种方式的读取 3.4G 大小文件的性能对比

图片

由此可见使用 JavaNIO FileChannnel 明显更优，但是 FileChannel 的方式是先读取固定大小缓冲区，不支持按行读取。也无法保证缓冲区正好包括整数行数据。如果缓冲区最后一个字节正好卡在一行数据中间，还需要额外配合读取下一批数据。如何把缓冲区变为一行行数据，比较困难。

File file = new File("/xxx.txt");
FileInputStream fileInputStream = null;
long now = System.currentTimeMillis();
try {
       fileInputStream = new FileInputStream(file);
       FileChannel fileChannel = fileInputStream.getChannel();

       int capacity = 1 * 1024 * 1024;//1M
       ByteBuffer byteBuffer = ByteBuffer.allocate(capacity);
       StringBuffer buffer = new StringBuffer();
       int size = 0;
       while (fileChannel.read(byteBuffer) != -1) {
          //读取后，将位置置为0，将limit置为容量, 以备下次读入到字节缓冲中，从0开始存储
          byteBuffer.clear();
          byte[] bytes = byteBuffer.array();
          size += bytes.length;
       }
       System.out.println("file size:" + size);
} catch (FileNotFoundException e) {
   e.printStackTrace();
} catch (IOException e) {
   e.printStackTrace();
} finally {
   //TODO close资源.
}
System.out.println("Time:" + (System.currentTimeMillis() - now));

JavaNIO 是基于缓冲区的，ByteBuffer 可转为 byte 数组，需要转为字符串，并且要处理按行截断。

但是 BufferedReader JavaIO 方式读取可以天然支持按行截断，况且性能还不错，10G 文件，大致只需要读取 30s，由于导入的整体瓶颈在写入部分，即便 30s 读取完，也不会影响整体性能。所以文件读取使用 BufferedReader 逐行读取。即方案 3

如果协调读文件任务和写数据库任务

这块比较混乱，请耐心看完。

100 个读取任务，每个任务读取一批数据，立即写入数据库是否可以呢？前面提到了由于数据库并发写入的瓶颈，无法满足 1 个库同时并发大批量写入 10 个表，所以 100 个任务同时写入数据库，势必导致每个库同时有 10 个表同时在顺序写，这加剧了磁盘的并发写压力。为尽可能提高速度，减少磁盘并发写入带来的性能下降，需要一部分写入任务被暂停的。那么读取任务需要限制并发度吗？不需要。

假设写入任务和读取任务合并，会影响读取任务并发度。初步计划读取任务和写入任务各自处理，谁也不耽误谁。但实际设计时发现这个方案较为困难。

最初的设想是引入 Kafka，即 100 个读取任务把数据投递到 Kafka，由写入任务消费 kafka 写入 DB。100 个读取任务把消息投递到 Kafka，此时顺序就被打乱了，如何保证有序写入数据库呢？我想到可以使用 Kafka partition 路由，即读取任务 id 把同一任务的消息都路由到同一个 partition，保证每个 partition 内有序消费。

要准备多少个分片呢？100 个很明显太多，如果 partition 小于 100 个，例如 10 个。那么势必存在多个任务的消息混合在一起。如果同一个库的多个表在一个 Kafka partition，且这个数据库只支持单表批量写入，不支持并发写多个表。这个库多个表的消息混在一个分片中，由于并发度的限制，不支持写入的表对应的消息只能被丢弃。所以这个方案既复杂，又难以实现。

所以最终放弃了 Kafka 方案，也暂时放弃了将读取和写入任务分离的方案。

最终方案简化为读取任务读一批数据，写入一批。即任务既负责读文件、又负责插入数据库。

如何保证任务的可靠性

如果读取任务进行到一半，宕机或者服务发布如何处理呢？或者数据库故障，一直写入失败，任务被暂时终止，如何保证任务再次拉起时，再断点处继续处理，不会存在重复写入呢？

刚才我们提到可以为每一个记录设置一个主键 Id，即文件后缀 index + 文件所在行号。可以通过主键 id 的方式保证写入的幂等。

文件所在的行号，最大值大致为 10G/1k = 10M，即 10000000。拼接最大的后缀 99。最大的 id 为 990000000。

所以也无需数据库自增主键 ID，可以在批量插入时指定主键 ID。

如果另一个任务也需要导入数据库呢？如何实现主键 ID 隔离，所以主键 ID 还是需要拼接 taskId。例如 {taskId}{fileIndex}{fileRowNumber} 转化为 Long 类型。如果 taskId 较大，拼接后的数值过大，转化为 Long 类型可能出错。

最重要的是，如果有的任务写入 1kw，有的其他任务写入 100W，使用 Long 类型无法获知每个占位符的长度，存在冲突的可能性。而如果拼接字符串 {taskId}{fileIndex}{fileRowNumber} ，新增唯一索引，会导致插入性能更差, 无法满足最快导入数据的诉求。所以需要想另一个方案。

可以考虑使用 Redis 记录当前任务的进度。例如 Redis 记录 task 的进度，批量写入数据库成功后，更新 task 进度。

INCRBY KEY_NAME INCR_AMOUNT

指定当前进度增加 100，例如 incrby task_offset_{taskId} 100。如果出现批量插入失败的，则重试插入。多次失败，则单个插入，单个更新 redis。要确保 Redis 更新成功，可以在 Redis 更新时也加上重试。

如果还不放心 Redis 进度和数据库更新的一致性，可以考虑消费数据库 binlog，每一条记录新增则 redis +1 。

如果任务出现中断，则首先查询任务的 offset。然后读取文件到指定的 offset 继续处理。

如何协调读取任务的并发度

前面提到了为了避免单个库插入表的并发度过高，影响数据库性能。可以考虑限制并发度。如何做到呢？

既然读取任务和写入任务合并一起。那么就需要同时限制读取任务。即每次只挑选一批读取写入任务执行。

在此之前需要设计一下任务表的存储模型。

图片

bizId 为了以后支持别的产品线，预设字段。默认为 1，代表当前业务线。
datbaseIndex 代表被分配的数据库后缀
tableIndex 代表被分配的表名后缀
parentTaskId，即总的任务 id
offset 可以用来记录当前任务的进度
10 亿条数据导入数据库，切分为 100 个任务后，会新增 100 个 taskId，分别处理一部分数据，即一个 10G 文件。
status 状态用来区分当前任务是否在执行，执行完成。

如何把任务分配给每一个节点，可以考虑抢占方式。每个任务节点都需要抢占任务，每个节点同时只能抢占 1 个任务。具体如何实现呢？可以考虑每个节点都启动一个定时任务，定期扫表，扫到待执行子任务，尝试执行该任务。

如何控制并发呢？可以使用 redission 的信号量。key 为数据库 id、

RedissonClient redissonClient = Redisson.create(config);
RSemaphore rSemaphore = redissonClient.getSemaphore("semaphore");
  // 设置1个并发度
rSemaphore.trySetPermits(1);
rSemaphore.tryAcquire();//申请加锁,非阻塞。

由任务负责定期轮训，抢到名额后，就开始执行任务。将该任务状态置为 Process，任务完成后或失败后，释放信号量。

图片

但是使用信号量限流有个问题，如果任务忘记释放信号量，或者进程 Crash 无法释放信号量，如何处理呢？可以考虑给信号量增加一个超时时间。那么如果任务执行过长，导致提前释放信号量，另一个客户单争抢到信号量，导致两个客户端同时写一个任务如何处理呢？

what，明明是将 10 亿数据导入数据库，怎么变成分布式锁超时的类似问题?

实际上 Redisson 的信号量并没有很好的办法解决信号量超时问题，正常思维：如果任务执行过长，导致信号量被释放，解决这个问题只需要续约就可以了，任务在执行中，只要发现快信号量过期了，就续约一段时间，始终保持信号量不过期。但是 Redission 并没有提供信号量续约的能力，怎么办？

不妨换个思路，我们一直在尝试让多个节点争抢信号量，进而限制并发度。可以试试选取一个主节点，通过主节点轮训任务表。分三种情况，

情况 1 当前执行中数量小于并发度。

则选取 id 最小的待执行任务，状态置为进行中，通知发布消息。
消费到消息的进程，申请分布式锁，开始处理任务。处理完成释放锁。借助于 Redission 分布式锁续约，保证任务完成前，锁不会超时。

情况 2 当前执行中数量等于并发度。

主节点尝试 get 进行中任务是否有锁。
如果没有锁，说明有任务执行失败，此时应该重新发布任务。如果有锁，说明有任务正在执行中。

情况 3 当前执行中数量大于并发度

上报异常情况，报警，人工介入

使用主节点轮训任务，可以减少任务的争抢，通过 kafka 发布消息，接收到消息的进程处理任务。为了保证更多的节点参与消费，可以考虑增加 Kafka 分片数。虽然每个节点可能同时处理多个任务，但是不会影响性能，因为性能瓶颈在数据库。

那么主节点应该如何选取呢？可以通过 Zookeeper+curator 选取主节点。可靠性比较高。

10 亿条数据插入数据库的时间影响因素非常多。包括数据库磁盘类型、性能。数据库分库数量如果能切分 1000 个库当然性能更快，要根据线上实际情况决策分库和分表数量，这极大程度决定了写入的速率。最后数据库批量插入的阈值也不是一成不变的，需要不断测试调整，以求得最佳的性能。可以按照 100,1000,10000 等不断尝试批量插入的最佳阈值。

最后总结一下几点重要的

总结

要首先确认约束条件，才能设计方案。确定面试官主要想问的方向，例如 1T 文件如何切割为小文件，虽是难点，然而可能不是面试官想考察的问题。
从数据规模看，需要分库分表，大致确定分表的规模。
从单库的写入瓶颈分析，判断需要进行分库。
考虑到磁盘对并发写的支持力度不同，同一个库多个表写入的并发需要限制。并且支持动态调整，方便在线上环境调试出最优值。
MySQL innodb、myisam 存储引擎对写入性能支持不同，也要在线上对比验证
数据库批量插入的最佳阈值需要反复测试得出。
由于存在并发度限制，所以基于 Kafka 分离读取任务和写入任务比较困难。所以合并读取任务和写入任务。
需要 Redis 记录任务执行的进度。任务失败后，重新导入时，记录进度，可避免数据重复问题。
分布式任务的协调工作是难点，使用 Redission 信号量无法解决超时续约问题。可以由主节点分配任务 + 分布式锁保证任务排他写入。主节点使用 Zookeeper+Curator 选取。

技术 #mysql

浏览 (574)