MySQL快速插入亿级数据

ゞ浴缸里的玫瑰 2023-10-04 17:32 86阅读 0赞

前言  
通常我们会有这样的需求：构造数据，大批量插入数据库，以供后续处理。如果是几万几十万的数据那还好说，但如果是上千万上亿的数据，那么我们对速度的追求就更加迫切。

这里我们利用Python来操作MySQL（或者PostgreSQL），采取多进程多协程的方式，实现大批量数据快速插入。

数据库配置

# 查看bulk_insert_buffer_size大小，默认是8M
    SHOW VARIABLES;
    # 查看最大链接数，默认可能是100
    show variables like 'max_connections';

根据需求适当更改配置，以centos7为例：

# 打开MySQL配置文件
    vim /etc/my.conf
    
    # 添加或者修改
    bulk_insert_buffer_size=120M
    
    max_connections=1000
    
    net_buffer_length=512k
    max_allowed_packet=100M

`这里要在数据库服务器上运行程序，host为127.0.0.1`

`如果是远程服务器要写上对应的服务器ip，不过这样的话速度就降到1000条每秒左右`

settings文件：

# MySQL数据库信息
    host = '127.0.0.1'
    user = 'root'
    password = '你的密码'
    port = 3306
    database = '数据库名称'

### 数据库链接 ###

定义一个专门用来操作数据库的类，创建文件mysql\_manager.py

# -*- coding: utf-8 -*-
    
    import pymysql
    from .settings import *
    
    
    class MysqlManager(object):
        """
        数据库操作
        """
        def __init__(self):
            self.user = user
            self.password = password
            self.database = database
            self.host = host
            self.port = port
            self.conn = None
            self.cur = None
    
        def connect_database(self):
            try:
                self.conn = pymysql.connect(self.host, self.user, self.password, self.database, self.port)
                print("connect database success")
            except Exception as err:
                print("connect database failed", err)
                return False
            self.cur = self.conn.cursor()
            return True
    
        def db_close(self):
            self.cur.close()
            self.conn.close()
    
        def db_commit(self):
            self.conn.commit()
    
        def execute(self, data, table):  # 保存到mysql
    
            keys = ', '.join(data.keys())
            values = ', '.join(['%s'] * len(data.keys()))
    
            sql = 'INSERT INTO {table}({keys}) VALUES ({values}) '.format(table=table, keys=keys, values=values)
    
            try:
                if self.cur:
                    # executemany批量插入
                    self.cur.executemany(sql, data)
                return True
            except Exception as e:
                print('Failed', e)
                if self.conn:
                    self.conn.rollback()
                return False

### 插入数据 ###

**构造字典数据那边，记得根据你自己创建的表结构，构造data数据。**

**插入数据那边，替换为你自己创建的数据库的表名。**

# -*- coding: utf-8 -*-
    
    import time
    import gevent
    from gevent import monkey
    import multiprocessing
    from .settings import *
    from mysql_manager import MysqlManager
    
    monkey.patch_all()  # 多协程gevent全部替换生效
    
    
    class MysqlInsert(object):
        def __init__(self):
            self.max_line = 0
            self.start_num = 0
            self.end_num = 0
    
        @staticmethod
        def run(my_min, my_max):
            # 实例化数据库相关操作
            db_handle = MysqlManager()
            # 链接数据库
            connect = db_handle.connect_database()
            if connect:
                data_many = []
                for i in range(my_min, my_max):
                    # 构造字典数据，这里根据你自己创建的表结构，构造data数据
                    data = {'a': i}
                    # 转化为元祖列表
                    data_many.append(tuple(data.values()))
                # 插入数据，指定表名
                insert_db = db_handle.execute(data_many, '表名')
    
                if insert_db:
                    print("successful，%s" % insert_db)
    
                    db_handle.db_commit()
                    db_handle.db_close()
                else:
                    print('Database insert failed')
    
        def thread_pool(self):
            max_line = self.max_line  # 定义每次插入多少条。
            # 这里的数据代表是每个进程插入多少，这里的数据除以max_line ,代表启动多少线程。
            g_l = [gevent.spawn(self.run, i, i + max_line) for i in range(self.start_num, self.end_num, max_line)]
            gevent.joinall(g_l)  # 等待所以操作都执行完毕
    
        def start_process(self):
            try:
                start_one_time = time.time()
                # 开启多线程
                p = multiprocessing.Process(target=self.thread_pool)
                p.start()
                p.join()
    
                one_time = time.time() - start_one_time
                print("一次耗时%s" % one_time)
                return one_time
            except Exception as e:
                print(e)
    
        def main(self):
            # 每次50万，循环10次，500万
            for i in range(0, 10):
                self.max_line = 5000  # 每次插入多少条
                self.end_num = 500000 * (i + 1)  # 最大值
                self.start_num = 1 + 500000 * i  # 初始值
    
                self.start_process()
    
    
    if __name__ == '__main__':
        start_time = time.time()
    
        test = MysqlInsert()
        test.main()
    
        all_time = time.time() - start_time
        print("总共耗时%s" % all_time)

### 结果 ###

![在这里插入图片描述][20191018172915748.png]

![在这里插入图片描述][20191018172925602.png]

最后我们可以看到，插入500万条数据总共耗时198秒，插入速度大概在25000条/每秒左右。其实我测试的时候构造的模拟数据是有调用数学计算方法的，会有一定的影响。当然你也可以调整main函数里面的参数，进一步提高插入速度。

代码借鉴了别人的方法，现在找不到链接了，整理的比较简陋。速度也还有提升的空间，但过犹不及，这个速度我目前还是能够接受的。

目前已经转移到PostgreSQL数据库上了，和MySQL相比两个是一样的，稍微改动一下就好了。数据表批量创建用Python来操作也很方便，大家应该都会。

[20191018172915748.png]: https://img-blog.csdnimg.cn/20191018172915748.png
[20191018172925602.png]: https://img-blog.csdnimg.cn/20191018172925602.png