python+mysql：实现python造1000万条数据并快速插入数据库，速度对比

速度对比"/>

python+mysql：实现python造1000万条数据并快速插入数据库，速度对比

要求：

• 造一千万条个人数据，包括：姓名、生日、身份证、证件类型、性别、民族等
• 每个人随机分配到20个组其中之一

实现思路：

• 思路一：用python生成1000万条数据并写入txt文件，再用pymysql写入数据库
• 思路二：用for循环不断写进变量，分批次插入数据库

实现过程：

思路一 很快就被否定了，因为在写入txt的文件过程中，随着数据量越来越大，速度成线性下降，导致代码跑了一晚上也没写完，更别说插入了，即使用load data也没法补救前期写数据的时间。

思路二 在实现的过程中变量随着数据堆积，速度也会越来越慢，也是成线性下降，

所以循环10000次后就执行一次插入：

for i in tqdm(range(begin, end)): # tqdm: python的进度条库if i % 10000 == 0:sql = sql.strip(',\n')try:db.ping(reconnect=True)cur = db.cursor()cur.execute(sql)dbmit()cur.close()db.close()del sql  # 删除变量gc.collect()  # 回收内存空间except Exception as e:print(e)

插入变量后直接删除变量，用python自带的gc.collect回收内存空间，这样可以提高下次变量赋值的速度；

结果：

单进程 : 

insertSQl(1, 10000001)

结果 耗时30:27 

100%|█████████████████████████████| 10000000/10000000 [30:27<00:00, 5472.67it/s]

多线程： 

class myThread(threading.Thread):def __init__(self, threadName, begin, end):threading.Thread.__init__(self)self.threadName = threadNameself.begin = beginself.end = enddef run(self) -> None:print(self.threadName + '开始执行')threadLock.acquire()insertSQl(self.threadName, self.begin, self.end)threadLock.release()print(self.threadName + '结束执行')if __name__ == '__main__':threadLock = threading.Lock()threads = []thread1 = myThread('thread1', 1, 1250001)thread2 = myThread('thread2', 1250001, 2500001)thread3 = myThread('thread3', 2500001, 3750001)thread4 = myThread('thread4', 3750001, 5000001)thread5 = myThread('thread5', 5000001, 6250001)thread6 = myThread('thread6', 6250001, 7500001)thread7 = myThread('thread7', 7500001, 8750001)thread8 = myThread('thread8', 8750001, 10000001)thread1.start()thread2.start()thread3.start()......threads.append(thread1)threads.append(thread2)threads.append(thread3)......for t in threads:t.join()

结果 耗时30:57，这是我没想到的，应该是我用错了，对线程不是很了解。。。

多进程：

startTime = datetime.datetime.now()
p = Pool(cpu_count())  # cpu_count 查询当前设备进程数，我的是八核，所以下面分了8个进程
p.apply_async(insertSQl, args=[1, 1250001])
p.apply_async(insertSQl, args=[1250001, 2500001])
p.apply_async(insertSQl, args=[2500001, 3750001])
p.apply_async(insertSQl, args=[3750001, 5000001])
p.apply_async(insertSQl, args=[5000001, 6250001])
p.apply_async(insertSQl, args=[6250001, 7500001])
p.apply_async(insertSQl, args=[7500001, 8750001])
p.apply_async(insertSQl, args=[8750001, 10000001])p.close()
p.join()
endTime = datetime.datetime.now()
print(endTime - startTime)

结果 耗时7:06

完整代码：

"""多进程"""
import datetime
import gc
from multiprocessing import cpu_count, Pool
import pymysql
from tqdm import tqdmdb = pymysql.Connect(host='ip地址',user='用户名',password='密码',database='数据库',port='端口号',charset='utf8',autocommit=True
)def insertSQl(begin, end):sql = 'sql语句'for i in tqdm(range(begin, end)):  # python进度条 tqdmif i % 10000 == 0:  # 每10000次进行一次插入，速度没有经过校验，不确定10000万次是否为最佳sql = sql.strip(',\n')  # sql语句拼接时最后会留下一个逗号，不删除会报错try:db.ping(reconnect=True)  # reconnect 自动重联cur = db.cursor()cur.execute(sql)dbmit()cur.close()db.close()del sql  # 删除变量gc.collect()  # 回收内存空间except Exception as e:print(e)# sql重置sql = 'sql语句'"""此处有个变量拼接，除余不满一万次时会继续拼接变量"""if __name__ == '__main__':startTime = datetime.datetime.now()p = Pool(cpu_count())  # cpu_count 查询当前设备进程数，我的是八核，所以下面分了8个进程p.apply_async(insertSQl, args=[1, 1250001])p.apply_async(insertSQl, args=[1250001, 2500001])p.apply_async(insertSQl, args=[2500001, 3750001])p.apply_async(insertSQl, args=[3750001, 5000001])p.apply_async(insertSQl, args=[5000001, 6250001])p.apply_async(insertSQl, args=[6250001, 7500001])p.apply_async(insertSQl, args=[7500001, 8750001])p.apply_async(insertSQl, args=[8750001, 10000001])p.close()p.join()endTime = datetime.datetime.now()print(endTime - startTime)

从结果看多进程的效率要比其他的好的多，翻了四五倍。水平有限，如果还有什么更好的办法，欢迎分享给我～～～