用Python并行處理大文件，看這篇就夠了！ - 網安 - 專業的網絡安全產業、社區、知識平臺

為了進行并行處理，我們將任務劃分為多個子單元。它增加了程序處理的工作數量，減少了整體處理時間。

例如，如果你正在處理一個大的CSV文件，你想修改一個單列。我們將把數據以數組的形式送入函數，它將根據可用的工作者的數量，一次并行處理多個值。這些工作器是基于你的處理器內的核心數量的。

注意：在一個較小的數據集上使用并行處理，不會提高處理時間。

在這篇博客中，我們將學習如何使用多處理、joblib和tqdm Python包減少大文件的處理時間。這是一個簡單的教程，可以適用于任何文件、數據庫、圖像、視頻和音頻。

注意：我們使用的是Kaggle筆記本進行實驗。處理時間可能因機器不同而不同。

開始處理

我們將使用Kaggle的美國事故（2016 - 2021）數據集（https://www.kaggle.com/datasets/sobhanmoosavi/us-accidents），該數據集由280萬條記錄和47列組成。

我們將導入multiprocessing、joblib和tqdm用于并行處理，pandas用于數據攝取，re、nltk和string用于文本處理。

# Parallel Computing
import multiprocessing as mp
from joblib import Parallel, delayed
from tqdm.notebook import tqdm
# Data Ingestion 
import pandas as pd
# Text Processing 
import re 
from nltk.corpus import stopwords
import string

在我們直接進入之前，讓我們通過加倍cpu_count()來設置n_workers。正如你所看到的，我們有8個工人。

n_workers = 2 * mp.cpu_count()
print(f"{n_workers} workers are available")
>>> 8 workers are available

在下一步，我們將使用pandas read_csv函數攝取大型CSV文件。然后，打印出數據框的形狀、列的名稱和處理時間。

注意：Jupyter的神奇函數%time可以在處理結束后顯示CPU times和wall time。

%%time
file_name="../input/us-accidents/US_Accidents_Dec21_updated.csv"
df = pd.read_csv(file_name)
print(f"Shape:{df.shape}\n\nColumn Names:\n{df.columns}\n")

輸出

Shape:(2845342, 47)
Column Names:
Index(['ID', 'Severity', 'Start_Time', 'End_Time', 'Start_Lat', 'Start_Lng',
'End_Lat', 'End_Lng', 'Distance(mi)', 'Description', 'Number', 'Street',
'Side', 'City', 'County', 'State', 'Zipcode', 'Country', 'Timezone',
'Airport_Code', 'Weather_Timestamp', 'Temperature(F)', 'Wind_Chill(F)',
'Humidity(%)', 'Pressure(in)', 'Visibility(mi)', 'Wind_Direction',
'Wind_Speed(mph)', 'Precipitation(in)', 'Weather_Condition', 'Amenity',
'Bump', 'Crossing', 'Give_Way', 'Junction', 'No_Exit', 'Railway',
'Roundabout', 'Station', 'Stop', 'Traffic_Calming', 'Traffic_Signal',
'Turning_Loop', 'Sunrise_Sunset', 'Civil_Twilight', 'Nautical_Twilight',
'Astronomical_Twilight'],
dtype='object')
CPU times: user 33.9 s, sys: 3.93 s, total: 37.9 s
Wall time: 46.9 s

清理文本

clean_text是一個用于處理和清理文本的簡單函數。我們將使用nltk.copus獲得英語停止詞，并使用它來過濾掉文本行中的停止詞。之后，我們將刪除句子中的特殊字符和多余的空格。這將是確定串行、并行和批處理的處理時間的基線函數。

def clean_text(text): 
  # Remove stop words
  stops = stopwords.words("english")
  text = " ".join([word for word in text.split() if word 
 not in stops])
  # Remove Special Characters
  text = text.translate(str.maketrans('', '', string.punctuation))
  # removing the extra spaces
  text = re.sub(' +',' ', text)
  return text

串行處理

對于串行處理，我們可以使用pandas的.apply()函數，但是如果你想看到進度條，你需要為pandas激活tqdm，然后使用.progress_apply()函數。

我們將處理280萬條記錄，并將結果保存回 "描述 "列。

%%time
tqdm.pandas()
df['Description'] = df['Description'].progress_apply(clean_text)

輸出

高端處理器串行處理280萬條記錄用了9分5秒。

100% ?????????? 2845342/2845342 [09:05<00:00, 5724.25it/s]
CPU times: user 8min 14s, sys: 53.6 s, total: 9min 7s
Wall time: 9min 5s

多處理

有多種方法可以對文件進行并行處理，我們將了解所有這些方法。multiprocessing是一個內置的python包，通常用于并行處理大文件。

我們將創建一個有8個工作者的多處理池，并使用map函數來啟動進程。為了顯示進度條，我們將使用tqdm。

map函數由兩部分組成。第一個部分需要函數，第二個部分需要一個參數或參數列表。

通過閱讀文檔了解更多。（https://docs.python.org/3/library/multiprocessing.html）

%%time
p = mp.Pool(n_workers) 
df['Description'] = p.map(clean_text,tqdm(df['Description']))

輸出

我們的處理時間幾乎提高了3倍。處理時間從9分5秒下降到3分51秒。

100% ?????????? 2845342/2845342 [02:58<00:00, 135646.12it/s]
CPU times: user 5.68 s, sys: 1.56 s, total: 7.23 s
Wall time: 3min 51s

并行處理

我們現在將學習另一個Python包來執行并行處理。在本節中，我們將使用joblib的Parallel和delayed來復制map函數。

Parallel需要兩個參數：n_job = 8和backend = multiprocessing。
然后，我們將在delayed函數中加入clean_text。
創建一個循環，每次送入一個值。下面的過程是相當通用的，你可以根據你的需要修改你的函數和數組。我曾用它來處理成千上萬的音頻和視頻文件，沒有任何問題。

建議：使用 "try: "和 "except: "添加異常處理。

def text_parallel_clean(array):
  result = Parallel(n_jobs=n_workers,backend="multiprocessing")(
  delayed(clean_text)
  (text) 
  for text in tqdm(array)
  )
  return result

在text_parallel_clean()中添加 "描述 "列。

%%time
df['Description'] = text_parallel_clean(df['Description'])

輸出

我們的函數比多處理Pool多花了13秒。即使如此，并行處理也比串行處理快4分59秒。

100% ?????????? 2845342/2845342 [04:03<00:00, 10514.98it/s]
CPU times: user 44.2 s, sys: 2.92 s, total: 47.1 s
Wall time: 4min 4s

并行批量處理

有一個更好的方法來處理大文件，就是把它們分成若干批，然后并行處理。讓我們從創建一個批處理函數開始，該函數將在單個批次的值上運行clean_function。

批量處理函數

def proc_batch(batch):
  return [
  clean_text(text)
  for text in batch
  ]

將文件分割成批處理

下面的函數將根據工作者的數量把文件分成多個批次。在我們的例子中，我們得到8個批次。

def batch_file(array,n_workers):
  file_len = len(array)
  batch_size = round(file_len / n_workers)
  batches = [
  array[ix:ix+batch_size]
  for ix in tqdm(range(0, file_len, batch_size))
  ]
  return batches
batches = batch_file(df['Description'],n_workers)
>>> 100% ?????????? 8/8 [00:00<00:00, 280.01it/s]

運行并行的批處理

最后，我們將使用Parallel和delayed來處理批次。

注意：為了得到一個單數組的值，我們必須運行列表理解，如下圖所示。

%%time
batch_output = Parallel(n_jobs=n_workers,backend="multiprocessing")(
  delayed(proc_batch)
  (batch) 
  for batch in tqdm(batches)
  )
df['Description'] = [j for i in batch_output for j in i]

輸出

我們已經改善了處理時間。這種技術在處理復雜數據和訓練深度學習模型方面非常有名。

100% ?????????? 8/8 [00:00<00:00, 2.19it/s]
CPU times: user 3.39 s, sys: 1.42 s, total: 4.81 s
Wall time: 3min 56s

tqdm 并發

tqdm將多進程帶到了一個新的水平。它簡單而強大。我將向每個數據科學家推薦它。

查看文檔，了解更多關于多處理的信息。（https://tqdm.github.io/）

process_map需要：

函數名稱
數據框架列
max_workers
chucksize與批量大小類似。我們將使用工人的數量來計算批處理的大小，或者你可以根據你的偏好來添加這個數字。

%%time
from tqdm.contrib.concurrent import process_map
batch = round(len(df)/n_workers)
df['Description'] = process_map(clean_text,df['Description'], max_workers=n_workers, chunksize=batch)

輸出

通過一行代碼，我們得到了最好的結果。

100% ?????????? 2845342/2845342 [03:48<00:00, 1426320.93it/s]
CPU times: user 7.32 s, sys: 1.97 s, total: 9.29 s
Wall time: 3min 51s

結論

你需要找到一個平衡點，選擇最適合你情況的技術。它可以是串行處理、并行處理或批處理。如果你正在處理一個較小的、不太復雜的數據集，并行處理可能會適得其反。

在這個迷你教程中，我們已經了解了各種Python包和技術，它們允許我們對數據函數進行并行處理。

如果你只是在處理一個表格數據集，并且想提高你的處理性能，那么我將建議你嘗試Dask（https://www.dask.org/）、datatable（https://github.com/h2oai/datatable）和RAPIDS（https://rapids.ai/）。

參考：https://www.kdnuggets.com/2022/07/parallel-processing-large-file-python.html