100%讓你在10分鐘內學會如何用python將數據批量的插入到數據庫

本文基於python, 使用pandas, pymysql等三方庫實現了向數據庫中高效批量插入數據，一方面提供被網上很多瞎轉載的答案給坑蒙了的人(因為我也是)，一方面自己也做個筆記，以後方便查閲

需求原因

最近在處理一個需求，有關批量往數據庫插入數據的，描述如下

• 原來的程序是基於sql的存儲過程進行數據的更新修改操作，由於數據量較大，導致對數據庫壓力太大，於是需要將程序重構為用python讀取文件的方式將數據做計算處理，減少這部分的壓力，最後僅僅將計算的結果調用aws的lambda服務重新更新到數據庫中就可以了，減少了極大的壓力，也降低了成本。涉及數據庫主要是插入及更新操作

版本庫信息邏輯梳理

實際上，最後一步，要寫入數據庫的文件數據是存儲在內存中的。因為讀取文件後進行的計算都是在內存中進行的，那麼計算的結果也沒必要再寫到本地，再去讀取，再寫入數據庫,這是會影響程序的效率的。邏輯如下

• 讀取文件
• 文件的拼接及計算，生成新的df
• 初始化數據庫的連接
• 將df所需數據轉換為元組數據(取決於數據庫的三方庫的接口是如何支持批量操作的)
• 將數據寫入數據庫
• 檢查數據庫內容即可

分步實現及分析

讀取文件

給文件路徑，然後去讀文件就行了，強調一下需要注意的點

• 絕對路徑: 這種最簡單，直接給路徑字符串就行了，但是一旦文件夾目錄結構變化，就需要頻繁的改
• 相對路徑: 我一般喜歡先在腳本中定位當前腳本的位置，然後通過相對路徑去找，這樣只要你整個包內部的目錄結構不變化，都不用改，就算部署上線也是直接根據包的位置來，很方便
• pandas默認會將所有數字讀取為float類型，所以對於那種看起來是數字，但實際上是需要當作字符串使用的字段進行類型的轉換

import pandas as pd import numpy as np# 當前腳本的位置current_folder_path = os.path.dirname(__file__)# 你的文件的位置your_file_path1 = os.path.join(current_folder_path, "文件的名字1")your_file_path2 = os.path.join(current_folder_path, "文件的名字2")# 我這裏是以讀取csv文件為例, delimiter為我們內部約定的列之間的分割符df1 = pd.read_csv(your_file_path1, dtype={"column1": str, "column2": str}, delimiter="/t")df2 = pd.read_csv(your_file_path2, dtype={"column1": str, "column2": str}, delimiter="/t")

文件的拼接及計算

文件的拼接主要就是merge和concat兩個語法的使用，強調一下小知識點

• merge語法主要是對應於sql語言的內連接，外連接，左連接和右連接等
• concat主要是用來將相同結構的df單純的拼接起來(也就是列表的總行數增加)

# 這裏以左連接舉例, 假設只有兩個文件拼接ret_df = pd.merge(df1, df2, left_on=["column_name"], right_on=["column_name"], how="left")

初始化連接

導入三方庫pymysql，初始化連接

# pymysql的接口獲取鏈接def mysql_conn(host, user, password, db, port=3306, charset="utf8"):  # 傳參版本  conn = pymysql.connect(host=host, user=user, password=password, database=db, port=port, charset=charset)  return conn

對應接口轉換數據

1. 數據插入要考慮寫入一個事務，因為失敗的話，要保證對數據庫沒有影響
2. 構造符合對應接口的數據格式，通過查詢，pymysql有兩種可以執行語句的接口
1. execute(單條插入語句)
1. 執行單條語句的接口
2. 類似這種: Insert into table_name (column) values (value);
2. executemany(批量插入語句)
1. 執行多條語句的接口
2. 類似這種: Insert into table_name (column1, column2, column3) values (value1, value2, value3);

具體實現如下

# 先創建cursor負責操作conn接口conn = mysql_conn("your db host", "your username", "your password", "db name")cursor = conn.cursor()# 開啓事務conn.begin()#############      構造批量數據的過程############## 先構造需要的或是和數據庫相匹配的列columns = list(df.columns)# 可以刪除不要的列或者數據庫沒有的列名columns.remove("列名")# 重新構造df,用上面的columns,到這裏你要保證你所有列都要準備往數據庫寫入了new_df = df[columns].copy()# 構造符合sql語句的列，因為sql語句是帶有逗號分隔的,(這個對應上面的sql語句的(column1, column2, column3))columns = ','.join(list(new_df.columns))# 構造每個列對應的數據，對應於上面的((value1, value2, value3))data_list = [tuple(i) for i in gdsord_df.values] # 每個元組都是一條數據，根據df行數生成多少元組數據# 計算一行有多少value值需要用字符串佔位s_count = len(data_list[0]) * "%s,"# 構造sql語句insert_sql = "insert into " + "數據庫表名" + " (" + columns + ") values (" + s_count[:-1] + ")"

將數據寫入數據庫

這個簡單，直接上代碼

cursor.executemany(insert_sql, data_list)conn.commit()cursor.close()conn.close()

檢查數據庫是否插入成功

如果沒問題的話，就可以同時進行多個文件讀寫，計算，最後啓用多線程同時向數據庫中寫入數據了，非常高效！

完整代碼

import pandas as pd import numpy as np# pymysql接口def mysql_conn(host, user, password, db, port=3306, charset="utf8"):  conn = pymysql.connect(host=host, user=user, password=password, database=db, port=port, charset=charset)  return conn# 當前腳本的位置current_folder_path = os.path.dirname(__file__)# 你的文件的位置your_file_path1 = os.path.join(current_folder_path, "文件的名字1")your_file_path2 = os.path.join(current_folder_path, "文件的名字2")# 我這裏是以讀取csv文件為例, delimiter為我們內部約定的列之間的分割符df1 = pd.read_csv(your_file_path1, dtype={"column1": str, "column2": str}, delimiter="/t")df2 = pd.read_csv(your_file_path2, dtype={"column1": str, "column2": str}, delimiter="/t")# 合併ret_df = pd.merge(df1, df2, left_on=["column_name"], right_on=["column_name"], how="left")# 先創建cursor負責操作conn接口conn = mysql_conn("your db host", "your username", "your password", "db name")cursor = conn.cursor()# 開啓事務conn.begin()# 先構造需要的或是和數據庫相匹配的列columns = list(df.columns)# 可以刪除不要的列或者數據庫沒有的列名columns.remove("列名")# 重新構造df,用上面的columns,到這裏你要保證你所有列都要準備往數據庫寫入了new_df = df[columns].copy()# 構造符合sql語句的列，因為sql語句是帶有逗號分隔的,(這個對應上面的sql語句的(column1, column2, column3))columns = ','.join(list(new_df.columns))# 構造每個列對應的數據，對應於上面的((value1, value2, value3))data_list = [tuple(i) for i in gdsord_df.values] # 每個元組都是一條數據，根據df行數生成多少元組數據# 計算一行有多少value值需要用字符串佔位s_count = len(data_list[0]) * "%s,"# 構造sql語句insert_sql = "insert into " + "數據庫表名" + " (" + columns + ") values (" + s_count[:-1] + ")"try:  cursor.executemany(insert_sql, data_list)  conn.commit()  cursor.close()  conn.close()except Exception as e:  # 萬一失敗了，要進行回滾操作  conn.rollback()  cursor.close()  conn.close()