自学教程：Python大批量写入数据(百万级别)的方法

import threadingimport pandas as pdimport asyncioimport timeimport aiomysqlimport pymysqldata=[]error_data=[]def run(start,end):    global data    global error_data    print("start"+threading.current_thread().name)    print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))    mysdb = getDb("*", *, "*", "*", "*")    cursor = mysdb.cursor()    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)"""    cursor.executemany(sql,data[start:end])    mysdb.commit()    mysdb.close()    print("end" + threading.current_thread().name)    print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))def csv_file_read_use_pd(csvFile):    csv_result = pd.read_csv(csvFile,encoding="utf-16",sep='/t')    csv_result = csv_result.fillna(value="None")    result = csv_result.values.tolist()    return resultclass MyDataBase:    def __init__(self,host=None,port=None,username=None,password=None,database=None):        self.db = pymysql.connect(host=host,port=port,user=username,password=password,database=database)    def close(self):        self.db.close()def getDb(host,port,username,password,database):    MyDb = MyDataBase(host, port, username, password,database)    return MyDb.dbdef main(csvFile):    global data  #获取全局对象  csv全量数据    #读取所有的数据   将所有数据均分成   thread_lens   份 分发给  thread_lens  个线程去执行    thread_lens=20    csv_result=csv_file_read_use_pd(csvFile)    day = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))    for item in csv_result:        item.insert(0,day)    data=csv_result    thread_exe_count_list=[]   #线程需要执行的区间    csv_lens=len(csv_result)    avg = csv_lens // thread_lens    remainder=csv_lens % thread_lens    # 0,27517  27517,55,034    nowIndex=0    for i in range(thread_lens):        temp=[nowIndex,nowIndex+avg]        nowIndex=nowIndex+avg        thread_exe_count_list.append(temp)    thread_exe_count_list[-1:][0][1]+=remainder  #余数分给最后一个线程    # print(thread_exe_count_list)    #th(thread_exe_count_list[0][0],thread_exe_count_list[0][1])    for i in range(thread_lens):        sub_thread = threading.Thread(target=run,args=(thread_exe_count_list[i][0],thread_exe_count_list[i][1],))        sub_thread.start()        sub_thread.join()        time.sleep(3)if __name__=="__main__":    #csv_file_read_use_pd("分公司箱型箱量.csv")    main("分公司箱型箱量.csv")

import threadingimport pandas as pdimport asyncioimport timeimport aiomysqldata=[]error_data=[]async def async_basic(loop,start,end):    global data    global error_data    print("start"+threading.current_thread().name)    print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))    conn = await aiomysql.connect(        host="*",        port=*,        user="*",        password="*",        db="*",        loop=loop    )    day = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))    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)"""    async with conn.cursor() as cursor:        for item in data[start:end]:            params=[day]            params.extend(item)            try:                x=await cursor.execute(sql,params)                if x==0:                    error_data.append(item)                print(threading.current_thread().name+"   result "+str(x))            except Exception as e:                print(e)                error_data.append(item)                time.sleep(10)                pass    await conn.close()    #await conn.commit()    #关闭连接池    # pool.close()    # await pool.wait_closed()    print("end" + threading.current_thread().name)    print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))def csv_file_read_use_pd(csvFile):    csv_result = pd.read_csv(csvFile,encoding="utf-16",sep='/t')    csv_result = csv_result.fillna(value="None")    result = csv_result.values.tolist()    return resultdef th(start,end):    loop = asyncio.new_event_loop()    loop.run_until_complete(async_basic(loop,start,end))def main(csvFile):    global data  #获取全局对象  csv全量数据    #读取所有的数据   将所有数据均分成   thread_lens   份 分发给  thread_lens  个线程去执行    thread_lens=20    csv_result=csv_file_read_use_pd(csvFile)    data=csv_result    thread_exe_count_list=[]   #线程需要执行的区间    csv_lens=len(csv_result)    avg = csv_lens // thread_lens    remainder=csv_lens % thread_lens    # 0,27517  27517,55,034    nowIndex=0    for i in range(thread_lens):        temp=[nowIndex,nowIndex+avg]        nowIndex=nowIndex+avg        thread_exe_count_list.append(temp)    thread_exe_count_list[-1:][0][1]+=remainder  #余数分给最后一个线程    print(thread_exe_count_list)    #th(thread_exe_count_list[0][0],thread_exe_count_list[0][1])    for i in range(thread_lens):        sub_thread = threading.Thread(target=th,args=(thread_exe_count_list[i][0],thread_exe_count_list[i][1],))        sub_thread.start()        time.sleep(3)if __name__=="__main__":    #csv_file_read_use_pd("分公司箱型箱量.csv")    main("分公司箱型箱量.csv")