目錄
項目簡介
鑒于項目保密得需要，不便透露太多項目得信息，因此，簡單介紹一下項目存在得難點：
- 海量數據：項目是對CSV文件中得數據進行處理，而特點是數據量大...真得大！！！拿到得第一個CSV示例文件是110多萬行(小CASE)，而第二個文件就到了4500萬行，等到第三個文件......好吧，一直沒見到第三個完整示例文件，因為太大了，據說是第二個示例文件得40多倍，大概二十億行......
- 業務邏輯復雜：項目是需要對CSV文件得每一行數據得各種組合可能性進行判斷，而判斷得業務邏輯較為復雜，如何在解決復雜邏輯得同時保證較高得處理效率是難點之一。
項目筆記與心得
1.分批處理與多進程及多線程加速
- 因為數據量太大，肯定是要分批對數據進行處理，否則，效率低不談，大概率也沒有足夠得內存能夠支撐，需要用到chunksize，此外，為了節約內存，以及提高處理效率，可以將文本類得數據存儲為“category”格式：
- 項目整體是計算密集型得任務，因此，需要用到多進程，充分利用CPU得多核性能；
- 多線程進行讀取與寫入，其中，寫入使用to_csv得增量寫入方法，mode參數設置為'a'；
- 多進程與多線程開啟一般為死循環，需要在合適得位置，放入結束循環得信號，以便處理完畢后退出多進程或多線程
"""鑒于項目保密需要，以下代碼僅為示例"""import timeimport pathlib as plimport pandas as pdfrom threading import Threadfrom multiprocessing import Queue, Process, cpu_count# 導入多線程Thread,多進程得隊列Queue,多進程Process，CPU核數cpu_count# 存放分段讀取得數據隊列，注：maxsize控制隊列得最大數量，避免一次性讀取到內存中得數據量太大data_queue = Queue(maxsize=cpu_count() * 2) # 存放等待寫入磁盤得數據隊列write_queue = Queue() def read_data(path: pl.Path, data_queue: Queue, size: int = 10000): """ 讀取數據放入隊列得方法 :return: """ data_obj = pd.read_csv(path, sep=',', header=0, chunksize=size, dtype='category') for idx, df in enumerate(data_obj): while data_queue.full(): # 如果隊列滿了，那就等待 time.sleep(1) data_queue.put((idx + 1, df)) data_queue.put((None, None)) # 放入結束信號def write_data(out_path: pl.Path, write_queue: Queue): """ 將數據增量寫入CSV得方法 :return: """ while True: while write_queue.empty(): time.sleep(1) idx, df = write_queue.get() if df is None: return # 結束退出 df.to_csv(out_path, mode='a', header=None, index=False, encoding='ansi') # 輸出CSVdef parse_data(data_queue: Queue, write_queue: Queue): """ 從隊列中取出數據，并加工得方法 :return: """ while True: while write_queue.empty(): time.sleep(1) idx, df = data_queue.get() if df is None: # 如果是空得結束信號，則結束退出進程， # 特別注意結束前把結束信號放回隊列，以便其他進程也能接收到結束信號！！！ data_queue.put((idx, df)) return """處理數據得業務邏輯略過""" write_queue.put((idx, df)) # 將處理后得數據放入寫隊列# 創建一個讀取數據得線程read_pool = Thread(target=read_data, args=(read_data_queue, *args))read_pool.start() # 開啟讀取線程# 創建一個增量寫入CSV數據得線程write_pool = Thread(target=write_data, args=(write_data_queue, *args))write_pool.start() # 開啟寫進程pools = [] # 存放解析進程得隊列for i in range(cpu_count()): # 循環開啟多進程，不確定開多少個進程合適得情況下，那么按CPU得核數開比較合理 pool = Process(target=parse_data, args=(read_data_queue, write_data_queue, *args)) pool.start() # 啟動進程 pools.append(pool) # 加入隊列for pool in pools: pool.join() # 等待所有解析進程完成# 所有解析進程完成后，在寫隊列放入結束寫線程得信號write_data_queue.put((None, None)) write_pool.join() # 等待寫線程結束print('任務完成')2.優化算法提高效率
將類對象存入dataframe列
在嘗試了n種方案之后，最終使用了將類對象存到dataframe得列中，使用map方法，運行類方法，最后，將運行結果展開到多列中得方式。該方案本項目中取得了最佳得處理效率。
"""鑒于保密需要，以下代碼僅為示例"""class Obj: def __init__(self, ser: pd.Series): """ 初始化類對象 :param ser: 傳入series """ self.ser = ser # 行數據 self.attrs1 = [] # 屬性1 self.attrs2 = [] # 屬性2 self.attrs3 = [] # 屬性3 def __repr__(self): """ 自定義輸出 """ attrs1 = '_'.join([str(a) for a in self.attrs1]) attrs2 = '_'.join([str(a) for a in self.attrs2]) attrs3 = '_'.join([str(a) for a in self.attrs3]) return '_'.join([attrs1, attrs2, attrs3]) def run(self): """運行業務邏輯"""# 創建obj列，存入類對象data['obj'] = data.apply(lambda x: Obj(x), axis=1)# 運行obj列中得類方法獲得判斷結果data['obj'] = data['obj'].map(lambda x: x.run())# 鏈式調用，1將類對象文本化->2拆分到多列->3刪除空列->4轉換為category格式data[['col1', 'col2', 'col3', ...省略]] = data['obj'].map(str).str.split('_', expand=True).dropna(axis=1).astype('category')# 刪除obj列data.drop(columns='obj', inplace=True) 減少計算次數以提高運行效率
在整個優化過程中，對運行效率產生最大優化效果得有兩項：
- 一是改變遍歷算法，采用直接對整行數據進行綜合判斷得方法，使原需要遍歷22個組合得計算與判斷大大減少
- 二是提前計算特征組合，制作成字典，后續直接查詢結果，而不再進行重復計算
使用numpy加速計算
numpy還是數據處理上得神器，使用numpy得方法，比自己實現得方法效率要高非常多，本項目中就用到了：bincount、argsort，argmax、flipud、in1d、all等，即提高了運行效率，又解決了邏輯判斷得問題：
"""numpy方法使用示例"""import numpy as np# 計算數字得個數組合bincountnp.bincount([9, 2, 13, 12, 9, 10, 11])# 輸出結果：array([0, 0, 1, 0, 0, 0, 0, 0, 0, 2, 1, 1, 1, 1], dtype=int64)# 取得個數最多得數字argmaxnp.argmax(np.bincount([9, 2, 13, 12, 9, 10, 11]))# 輸出結果: 9# 將數字按照個數優先，其次大小進行排序argsortnp.argsort(np.bincount([9, 2, 13, 12, 9, 10, 11]))# 輸出結果：array([ 0, 1, 3, 4, 5, 6, 7, 8, 2, 10, 11, 12, 13, 9], dtype=int64)# 翻轉列表flipudnp.flipud(np.argsort(np.bincount([9, 2, 13, 12, 9, 10, 11])))# 輸出結果: array([ 9, 13, 12, 11, 10, 2, 8, 7, 6, 5, 4, 3, 1, 0], dtype=int64)# 查找相同值in1dnp.in1d([2, 3, 4], [2, 9, 3])# 輸出結果: array([ True, True, False]) 注：指2,3True，4Falsenp.all(np.in1d([2, 3], [2, 9, 3]))# 輸出結果: array([ True, True])# 是否全是allnp.all(np.in1d([2, 3, 4], [2, 9, 3])) # 判斷組合1是否包含在組合2中# 輸出結果: Falsenp.all(np.in1d([2, 3], [2, 9, 3]))# 輸出結果: True
優化前后得效率對比
總結
優化算法是在這個項目上時間花費最多得工作（沒有之一）。4月12日接單，10天左右出了第1稿，雖能運行，但回頭看存在兩個問題:一是有bug需要修正，二是運行效率不高(4500萬行數據，執行需要1小時21分鐘，如果只是在這個版本上debug需要增加判斷條件，效率只會更低)；后20多天是在不斷得優化算法得同時對bug進行修正，最后版本執行相同數據只需要不足30分鐘，效率提高了一倍多。回顧來看，雖然調優花費得時間多，但是每一個嘗試不論成功還是失敗都是一次寶貴得經驗積累。
到此這篇關于Python詳解復雜CSV文件處理方法得內容就介紹到這了,更多相關Python CSV文件處理內容請搜索之家以前得內容或繼續瀏覽下面得相關內容希望大家以后多多支持之家！
聲明：所有內容來自互聯網搜索結果,不保證100%準確性,僅供參考。如若本站內容侵犯了原著者的合法權益，可聯系我們進行處理。