使用Python實現將多表分批次從數據庫導出到Excel

一、應用場景

為了避免反覆的手手工從後臺數據庫導出某些數據表到Excel文件、高效率到多份離線數據。

二、功能事項

支持一次性導出多個數據源表、自動獲取各表的字段名。

支持控制批次的寫入速率。例如：每5000行一個批次寫入到excel。

支持結構相同的表導入到同一個Excel文件。可適用於經過水平切分後的分佈式表。

三、主要實現

1、概覽

A[創建類] -->|方法1| B(創建數據庫連接)
A[創建類] -->|方法2| C(取查詢結果集)
A[創建類] -->|方法3| D(利用句柄寫入Excel)
A[創建類] -->|方法4| E(讀取多個源表)

B(創建數據庫連接) -->U(調用示例)
C(取查詢結果集) -->U(調用示例)
D(利用句柄寫入Excel) -->U(調用示例)
E(讀取多個源表) -->U(調用示例)

2、主要方法

首先需要安裝第三方庫pymssql實現對SQLServer的連接訪問，自定義方法__getConn()需要指定如下五個參數：服務器host、登錄用戶名user、登錄密碼pwd、指定的數據庫db、字符編碼charset。連接成功後，通過cursor()獲取遊標對象，它將用來執行數據庫腳本，並得到返回結果集和數據總量。

創建數據庫連接和執行SQL的源碼：

 def __init__(self,host,user,pwd,db): self.host = host self.user = user self.pwd = pwd self.db = db def __getConn(self): if not self.db: raise(NameError,'沒有設置數據庫信息') self.cOnn= pymssql.connect(host=self.host, user=self.user, password=self.pwd, database=self.db, charset='utf8') cur = self.conn.cursor() if not cur: raise(NameError,'連接數據庫失敗') else: return cur

3、方法3中寫入Excel時，注意一定要用到Pandas中的公共句柄ExcelWriter對象writer。當數據被分批多次寫入同一個文件時，如果直接使用to_excel()方法，則前面批次的結果集將會被後續結果覆蓋。增加了這個公共句柄限制後，後面的寫入會累加到前面寫入的數據尾部行，而不是全部覆蓋。

writer = pd.ExcelWriter(file)
df_fetch_data[rs_startrow:i*N].to_excel(writer, header=isHeader, index=False, startrow=startRow)

分批次寫入到目標Excel時的另一個要注意的參數是寫入行startrow的設置。每次寫入完成後需要重新指下一批次數據的初始位置值。每個批次的數據會記錄各自的所屬批次信息。

利用關鍵字參數**args 指定多個數據源表和數據庫連接。

 def exportToExcel(self, **args): for sourceTB in args['sourceTB']: arc_dict = dict( sourceTB = sourceTB, path=args['path'], startRow=args['startRow'], isHeader=args['isHeader'], batch=args['batch'] ) print(' 當前導出的數據表為：%s' %(sourceTB)) self.writeToExcel(**arc_dict) return 'success'

四、先用類MSSQL創建對象，再定義關鍵字參數args，最終調用方法導出到文件即完成數據導出。

 #!/usr/bin/env python # coding: utf-8 # 主要功能：分批次導出大數據量、結構相同的數據表到excel # 導出多個表的數據到各自的文件， # 目前問題：to_excel 雖然設置了分批寫入，但先前的數據會被下一次寫入覆蓋， # 利用Pandas包中的ExcelWriter()方法增加一個公共句柄，在寫入新的數據之時保留原來寫入的數據，等到把所有的數據都寫進去之後關閉這個句柄 import pymssql import pandas as pd import datetime import math class MSSQL(object): def __init__(self,host,user,pwd,db): self.host = host self.user = user self.pwd = pwd self.db = db def __getConn(self): if not self.db: raise(NameError,'沒有設置數據庫信息') self.cOnn= pymssql.connect(host=self.host, user=self.user, password=self.pwd, database=self.db, charset='utf8') cur = self.conn.cursor() if not cur: raise(NameError,'連接數據庫失敗') else: return cur def executeQuery(self,sql): cur = self.__getConn() cur.execute(sql) # 獲取所有數據集 # fetchall()獲取結果集中的剩下的所有行 # 如果數據量太大，是否需要分批插入 resList, rowcount = cur.fetchall(),cur.rowcount self.conn.close() return (resList, rowcount) # 導出單個數據表到excel def writeToExcel(self,**args): sourceTB = args['sourceTB'] columns = args.get('columns') path=args['path'] fname=args.get('fname') startRow=args['startRow'] isHeader=args['isHeader'] N=args['batch'] # 獲取指定源數據列 if columns is None: columns_select = ' * ' else: columns_select = ','.join(columns) if fname is None: fname=sourceTB+'_exportData.xlsx' file = path + fname # 增加一個公共句柄，寫入新數據時，保留原數據 writer = pd.ExcelWriter(file) sql_select = 'select '+ columns_select + ' from '+ sourceTB fetch_data, rowcount = self.executeQuery(sql_select) # print(rowcount) df_fetch_data = pd.DataFrame(fetch_data) # 一共有roucount行數據，每N行一個batch提交寫入到excel times = math.floor(rowcount/N) i = 1 rs_startrow = 0 # 當總數據量 > 每批插入的數據量時 print(i, times) is_while=0 while i <= times: is_while = 1 # 如果是首次，且指定輸入標題，則有標題 if i==1: # isHeader = True startRow = 1 else: # isHeader = False startRow+=N # 切片取指定的每個批次的數據行 ,前閉後開 # startrow: 寫入到目標文件的起始行。0表示第1行，1表示第2行。。。 df_fetch_data['batch'] = 'batch'+str(i) df_fetch_data[rs_startrow:i*N].to_excel(writer, header=isHeader, index=False, startrow=startRow) print('第',str(i),'次循環，取源數據第',rs_startrow,'行至',i*N,'行','寫入到第',startRow,'行') print('第',str(i),'次寫入數據為：',df_fetch_data[rs_startrow:i*N]) # 重新指定源數據的讀取起始行 rs_startrow =i * N i+=1 # 寫入文件的開始行數 # 當沒有做任何循環時，仍然從第一行開始寫入 if is_while == 0: startRow = startRow else: startRow+=N df_fetch_data['batch'] = 'batch'+str(i) print('第{0}次讀取數據，從第{1}行開始，寫入到第{2}行！'.format(str(i), str(rs_startrow), str(startRow))) print('第',str(i),'寫入數據為：',df_fetch_data[rs_startrow:i*N]) df_fetch_data[rs_startrow:i*N].to_excel(writer, header=isHeader, index=False, startrow=startRow) # 注： 這裡一定要saver()將數據從緩存寫入磁盤！！！！！！！！！！！！！！！！！！！！！1 writer.save() start_time=datetime.datetime.now() # 導出結構相同的多個表到同一樣excel def exportToExcel(self, **args): for sourceTB in args['sourceTB']: arc_dict = dict( sourceTB = sourceTB, path=args['path'], startRow=args['startRow'], isHeader=args['isHeader'], batch=args['batch'] ) print(' 當前導出的數據表為：%s' %(sourceTB)) self.writeToExcel(**arc_dict) return 'success' start_time=datetime.datetime.now() if __name__ == "__main__": ms = MSSQL(host="localhost",user="test",pwd="test",db="db_jun") args = dict( sourceTB = ['tb2', 'tb1'],# 待導出的表 path='D:myPCPython',# 導出到指定路徑 startRow=1,#設定寫入文件的首行，第2行為數據首行 isHeader=False,# 是否包含源數據的標題 batch=5 ) # 導出多個文件 ms.exportToExcel(**args)