Python大批量写入数据(百万级别)的方法

李万根

背景

现有一个百万行数据的csv格式文件，需要在两分钟之内存入数据库。

方案

方案一：多线程+协程+异步MySql方案二：多线程+MySql批量插入

代码

1.     1，先通过pandas读取所有csv数据存入列表。    2，设置N个线程，将一百万数据均分为N份，以start,end传递给线程以切片的方法读取区间数据（建议为16个线程）    3，方案二 线程内以  executemany 方法批量插入所有数据。    4，方案一 线程内使用异步事件循环遍历所有数据异步插入。     5，方案一纯属没事找事型。

复制代码

方案二

1. import threading
   
2. 
   
3. import pandas as pd
   
4. import asyncio
   
5. import time
   
6. 
   
7. import aiomysql
   
8. import pymysql
   
9. 
   
10. data=[]
    
11. error_data=[]
    
12. 
    
13. def run(start,end):
    
14.     global data
    
15.     global error_data
    
16.     print("start"+threading.current_thread().name)
    
17.     print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
    
18.     mysdb = getDb("*", *, "*", "*", "*")
    
19.     cursor = mysdb.cursor()
    
20.     sql = """insert into *_*_* values(%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)"""
    
21.     cursor.executemany(sql,data[start:end])
    
22.     mysdb.commit()
    
23.     mysdb.close()
    
24.     print("end" + threading.current_thread().name)
    
25.     print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
    
26. 
    
27. def csv_file_read_use_pd(csvFile):
    
28.     csv_result = pd.read_csv(csvFile,encoding="utf-16",sep='\t')
    
29.     csv_result = csv_result.fillna(value="None")
    
30.     result = csv_result.values.tolist()
    
31.     return result
    
32. 
    
33. class MyDataBase:
    
34.     def __init__(self,host=None,port=None,username=None,password=None,database=None):
    
35.         self.db = pymysql.connect(host=host,port=port,user=username,password=password,database=database)
    
36.     def close(self):
    
37.         self.db.close()
    
38. 
    
39. def getDb(host,port,username,password,database):
    
40.     MyDb = MyDataBase(host, port, username, password,database)
    
41.     return MyDb.db
    
42. 
    
43. def main(csvFile):
    
44.     global data  #获取全局对象  csv全量数据
    
45.     #读取所有的数据   将所有数据均分成   thread_lens   份 分发给  thread_lens  个线程去执行
    
46.     thread_lens=20
    
47.     csv_result=csv_file_read_use_pd(csvFile)
    
48.     day = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
    
49.     for item in csv_result:
    
50.         item.insert(0,day)
    
51. 
    
52.     data=csv_result
    
53.     thread_exe_count_list=[]   #线程需要执行的区间
    
54.     csv_lens=len(csv_result)
    
55.     avg = csv_lens // thread_lens
    
56.     remainder=csv_lens % thread_lens
    
57.     # 0,27517  27517,55,034
    
58.     nowIndex=0
    
59.     for i in range(thread_lens):
    
60.         temp=[nowIndex,nowIndex+avg]
    
61.         nowIndex=nowIndex+avg
    
62.         thread_exe_count_list.append(temp)
    
63.     thread_exe_count_list[-1:][0][1]+=remainder  #余数分给最后一个线程
    
64.     # print(thread_exe_count_list)
    
65. 
    
66.     #th(thread_exe_count_list[0][0],thread_exe_count_list[0][1])
    
67. 
    
68.     for i in range(thread_lens):
    
69.         sub_thread = threading.Thread(target=run,args=(thread_exe_count_list[i][0],thread_exe_count_list[i][1],))
    
70.         sub_thread.start()
    
71.         sub_thread.join()
    
72.         time.sleep(3)
    
73. 
    
74. if __name__=="__main__":
    
75.     #csv_file_read_use_pd("分公司箱型箱量.csv")
    
76.     main("分公司箱型箱量.csv")

复制代码

方案一

1. import threading
   
2. 
   
3. import pandas as pd
   
4. import asyncio
   
5. import time
   
6. 
   
7. import aiomysql
   
8. 
   
9. data=[]
   
10. error_data=[]
    
11. 
    
12. async def async_basic(loop,start,end):
    
13.     global data
    
14.     global error_data
    
15.     print("start"+threading.current_thread().name)
    
16.     print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
    
17.     conn = await aiomysql.connect(
    
18.         host="*",
    
19.         port=*,
    
20.         user="*",
    
21.         password="*",
    
22.         db="*",
    
23.         loop=loop
    
24.     )
    
25.     day = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
    
26.     sql = """insert into **** values(%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)"""
    
27.     async with conn.cursor() as cursor:
    
28.         for item in data[start:end]:
    
29.             params=[day]
    
30.             params.extend(item)
    
31.             try:
    
32.                 x=await cursor.execute(sql,params)
    
33.                 if x==0:
    
34.                     error_data.append(item)
    
35.                 print(threading.current_thread().name+"   result "+str(x))
    
36.             except Exception as e:
    
37.                 print(e)
    
38.                 error_data.append(item)
    
39.                 time.sleep(10)
    
40.                 pass
    
41.     await conn.close()
    
42.     #await conn.commit()
    
43.     #关闭连接池
    
44.     # pool.close()
    
45.     # await pool.wait_closed()
    
46.     print("end" + threading.current_thread().name)
    
47.     print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
    
48. 
    
49. def csv_file_read_use_pd(csvFile):
    
50.     csv_result = pd.read_csv(csvFile,encoding="utf-16",sep='\t')
    
51.     csv_result = csv_result.fillna(value="None")
    
52.     result = csv_result.values.tolist()
    
53.     return result
    
54. 
    
55. def th(start,end):
    
56.     loop = asyncio.new_event_loop()
    
57.     loop.run_until_complete(async_basic(loop,start,end))
    
58. 
    
59. 
    
60. def main(csvFile):
    
61.     global data  #获取全局对象  csv全量数据
    
62.     #读取所有的数据   将所有数据均分成   thread_lens   份 分发给  thread_lens  个线程去执行
    
63.     thread_lens=20
    
64.     csv_result=csv_file_read_use_pd(csvFile)
    
65.     data=csv_result
    
66.     thread_exe_count_list=[]   #线程需要执行的区间
    
67.     csv_lens=len(csv_result)
    
68.     avg = csv_lens // thread_lens
    
69.     remainder=csv_lens % thread_lens
    
70.     # 0,27517  27517,55,034
    
71.     nowIndex=0
    
72.     for i in range(thread_lens):
    
73.         temp=[nowIndex,nowIndex+avg]
    
74.         nowIndex=nowIndex+avg
    
75.         thread_exe_count_list.append(temp)
    
76.     thread_exe_count_list[-1:][0][1]+=remainder  #余数分给最后一个线程
    
77.     print(thread_exe_count_list)
    
78. 
    
79.     #th(thread_exe_count_list[0][0],thread_exe_count_list[0][1])
    
80. 
    
81.     for i in range(thread_lens):
    
82.         sub_thread = threading.Thread(target=th,args=(thread_exe_count_list[i][0],thread_exe_count_list[i][1],))
    
83.         sub_thread.start()
    
84.         time.sleep(3)
    
85. 
    
86. if __name__=="__main__":
    
87.     #csv_file_read_use_pd("分公司箱型箱量.csv")
    
88.     main("分公司箱型箱量.csv")

复制代码

总结

到此这篇关于Python大批量写入数据的文章就介绍到这了,更多相关Python大批量写入数据内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

来源:https://www.jb51.net/python/292125st5.htm
免责声明：由于采集信息均来自互联网，如果侵犯了您的权益，请联系我们【E-Mail:cb@itdo.tech】 我们会及时删除侵权内容，谢谢合作！