100%讓你在10分鐘內學會如何用Python將數據批量的插入到數據庫
我是一名掙扎在編程鏈底端的pythoner,工作中既要和數據打交道,也要保持和erp系統,web網站友好的"溝通"···,我會時不時的分享下工作中遇到那點事,包括個人覺得值得記錄的編程小技巧,還有就是遇到的問題以及解決方案,還有源碼的閱讀等等,可能也有編程中的生活感悟,不說了,我要去重構我的程序了

本文基于python, 使用pandas, pymysql等三方庫實現了向數據庫中高效批量插入數據,一方面提供被網上很多瞎轉載的答案給坑蒙了的人(因為我也是),一方面自己也做個筆記,以后方便查閱
需求原因
最近在處理一個需求,有關批量往數據庫插入數據的,描述如下
- 原來的程序是基于sql的存儲過程進行數據的更新修改操作,由于數據量較大,導致對數據庫壓力太大,于是需要將程序重構為用python讀取文件的方式將數據做計算處理,減少這部分的壓力,最后僅僅將計算的結果調用aws的lambda服務重新更新到數據庫中就可以了,減少了極大的壓力,也降低了成本。涉及數據庫主要是插入及更新操作
版本庫信息
- 基于linux系統寫的
- 三方庫 >>> pandas 1.0.5, pymysql 0.9.3
- python版本 >>> 3.7
- 標準庫 >> os
邏輯梳理
實際上,最后一步,要寫入數據庫的文件數據是存儲在內存中的。因為讀取文件后進行的計算都是在內存中進行的,那么計算的結果也沒必要再寫到本地,再去讀取,再寫入數據庫,這是會影響程序的效率的。邏輯如下
- 讀取文件
- 文件的拼接及計算,生成新的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
對應接口轉換數據
- 數據插入要考慮寫入一個事務,因為失敗的話,要保證對數據庫沒有影響
- 構造符合對應接口的數據格式,通過查詢,pymysql有兩種可以執行語句的接口
- execute(單條插入語句)
- 執行單條語句的接口
- 類似這種: Insert into table_name (column) values (value);
- executemany(批量插入語句)
- 執行多條語句的接口
- 類似這種: 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()