快捷導(dǎo)航

Python大批量寫入數(shù)據(jù)(百萬級(jí)別)的方法

更新時(shí)間：2023年07月14日 10:15:15 作者：西紅市杰出青年

這篇文章主要給大家介紹了關(guān)于Python大批量寫入數(shù)據(jù)(百萬級(jí)別)的相關(guān)資料,在日常處理數(shù)據(jù)的過程中,我們都有批量寫入數(shù)據(jù)的需求,文中給出了詳細(xì)的示例代碼,需要的朋友可以參考下

背景

現(xiàn)有一個(gè)百萬行數(shù)據(jù)的csv格式文件，需要在兩分鐘之內(nèi)存入數(shù)據(jù)庫。

方案

方案一：多線程+協(xié)程+異步MySql方案二：多線程+MySql批量插入

代碼

   1，先通過pandas讀取所有csv數(shù)據(jù)存入列表。
   2，設(shè)置N個(gè)線程，將一百萬數(shù)據(jù)均分為N份，以start,end傳遞給線程以切片的方法讀取區(qū)間數(shù)據(jù)（建議為16個(gè)線程）
   3，方案二線程內(nèi)以 executemany 方法批量插入所有數(shù)據(jù)。
   4，方案一線程內(nèi)使用異步事件循環(huán)遍歷所有數(shù)據(jù)異步插入。
   5，方案一純屬?zèng)]事找事型。

方案二

import threading

import pandas as pd
import asyncio
import time

import aiomysql
import pymysql

data=[]
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 result

class 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.db

def main(csvFile):
    global data  #獲取全局對(duì)象  csv全量數(shù)據(jù)
    #讀取所有的數(shù)據(jù)   將所有數(shù)據(jù)均分成   thread_lens   份 分發(fā)給  thread_lens  個(gè)線程去執(zhí)行
    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=[]   #線程需要執(zhí)行的區(qū)間
    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  #余數(shù)分給最后一個(gè)線程
    # 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 threading

import pandas as pd
import asyncio
import time

import aiomysql

data=[]
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()
    #關(guān)閉連接池
    # 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 result

def th(start,end):
    loop = asyncio.new_event_loop()
    loop.run_until_complete(async_basic(loop,start,end))


def main(csvFile):
    global data  #獲取全局對(duì)象  csv全量數(shù)據(jù)
    #讀取所有的數(shù)據(jù)   將所有數(shù)據(jù)均分成   thread_lens   份 分發(fā)給  thread_lens  個(gè)線程去執(zhí)行
    thread_lens=20
    csv_result=csv_file_read_use_pd(csvFile)
    data=csv_result
    thread_exe_count_list=[]   #線程需要執(zhí)行的區(qū)間
    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  #余數(shù)分給最后一個(gè)線程
    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")