所以我有大量需要處理成 CSV 的檔案。每個檔案本身都很大，每一行都是一個字串。檔案的每一行都可以代表三種資料中的一種，每種資料的處理方式都略有不同。我當前的解決方案如下所示：
type1_columns = [...]
type2_columns = [...]
type3_columns = [...]
file_list = os.listdir(filelist)
def process_type1_line(json_line):
#processing logic
to_append = [a, b, c, d, e]
type1_series = pd.Series(to_append, index=type1_columns)
return type1_series
def process_type2_line(json_line):
#processing logic
to_append = [a, b, c, d, e]
type2_series = pd.Series(to_append, index=type2_columns)
return type2_series
def process_type3_line(json_line):
#processing logic
to_append = [a, b, c, d, e]
type3_series = pd.Series(to_append, index=type3_columns)
return type3_series
def process_file(file):
type1_df = pd.DataFrame(columns=type1_columns)
type2_df = pd.DataFrame(columns=type2_columns)
type3_df = pd.DataFrame(columns=type3_columns)
with open(filepath/file) as f:
data=f.readlines()
for line in data:
#some logic to get the record_type and convert line to json
record_type = ...
json_line = ...
if record_type == "type1":
type1_series = process_type1_line(json_line)
type1_df = type1_df.append(type1_series, ignore_index=True)
if record_type == "type2":
type2_series = process_type2_line(json_line)
type2_df = type2_df.append(type2_series, ignore_index=True)
if record_type == "type3":
type3_series = process_type3_line(json_line)
type3_df = type3_df.append(type3_series, ignore_index=True)
type1_df.to_csv(type1_csv_path.csv)
type2_df.to_csv(type2_csv_path.csv)
type3_df.to_csv(type3_csv_path.csv)
for file in file_list:
process_file(file)
我遍歷檔案，并為三種不同型別的記錄中的每一種創建資料框。我決議這些行并為每個行呼叫適當的處理函式。回傳的系列將附加到該檔案的該記錄型別的最終資料幀。處理完檔案后，三個資料框將保存為 CSV，然后我們從下一個檔案開始。
問題是這種方法花費的時間太長，我需要數周時間來處理所有檔案。
我嘗試通過使用多處理（我沒有大量經驗）來修改我的方法，其中包括：
with ThreadPoolExecutor(max_workers=30) as executor:
futures = [executor.submit(process_file, file) for file in file_list]
在一些日志列印陳述句中，我可以看到這開始處理 30 個檔案，但沒有一個檔案完成，所以我至少知道我的方法有缺陷。誰能解釋解決這個問題的最佳方法是什么？也許是多處理和異步的某種組合？
uj5u.com熱心網友回復：
你有兩個大問題：
您將整個輸入檔案加載到記憶體中，在記憶體中生成整個結果，然后一次寫入整個輸出檔案。這意味著，如果您有 30 個并行操作的作業人員，您需要與 30 個（自我描述的）大檔案成比例的記憶體。您將所有資料存盤兩次，一次作為s回傳
list的行，然后再次存盤在三個s 之一中；如果您按原樣使用您的代碼，沒有執行程式，并且只是更改：strf.readlines()DataFramedata=f.readlines() for line in data:到：
for line in f:您會立即將記憶體使用量減少大約一半，這（可能）足以阻止頁面抖動。也就是說，您仍然會使用與檔案大小成比例的記憶體來存盤
DataFrames，因此如果您并行化您的代碼，您將繼續顛簸，并且如果檔案足夠大，即使沒有并行性也可能仍然顛簸。您正在
.append為每一行使用，IIRC，對于DataFrames 是 Schlemiel the Painter 演算法的一種形式：每個都append制作一個全新的DataFrame，將舊的全部內容DataFrame加上少量的新資料復制到一個新的DataFrame，用隨著現有資料越來越大，作業時間越來越長；應該攤銷O(n)的作業變成了O(n**2)作業。
在這兩者之間，您使用的記憶體比需要的多，并且在重復的追加上執行了大量不必要的忙碌作業。并行性可能有助于更快地完成繁忙的作業，但作為交換，它會將您的記憶體需求增加 30 倍；很可能，您沒有那么多 RAM（如果這些檔案真的很大，那么您可能沒有足夠的 RAM 來存盤其中一個檔案），并且您最終會出現頁面抖動（將記憶體寫入頁面檔案/交換檔案以為其他東西騰出空間，按需讀回，并經常丟棄在你完成之前分頁的記憶體，使記憶體訪問與磁盤性能相關，這比 RAM 訪問慢幾個數量級）。
我不太了解 Pandas，不能說它是否為你正在做的事情提供了一些更好的增量解決方案，但你并不真的需要它；只需逐行處理輸入，并使用csv模塊隨時寫入行。您的記憶體需求將從“與每個輸入檔案的大小成比例”下降到“與輸入檔案每一行的資料成比例”。
你的process_file函式最終看起來像：
def process_file(file):
# Open input file and all output files (newline='' needed to play nice with csv module
# which takes control of newline format to ensure dialect rules followed precisely,
# regardless of OS line separator rules)
with open(filepath/file) as f,\
open(type1_csv_path, 'w', newline='') as type1f,\
open(type2_csv_path, 'w', newline='') as type2f,\
open(type3_csv_path, 'w', newline='') as type3f:
csv1 = csv.writer(type1f)
csv1.writerow(type1_columns) # Omit if no useful column header
csv2 = csv.writer(type2f)
csv2.writerow(type2_columns) # Omit if no useful column header
csv3 = csv.writer(type3f)
csv3.writerow(type3_columns) # Omit if no useful column header
for line in f: # Directly iterating file object lazily fetches line at a time
# where .readlines() eagerly fetches whole file, requiring
# a ton of memory for no reason
#some logic to get the record_type and convert line to json
record_type = ...
json_line = ...
if record_type == "type1":
type1_series = process_type1_line(json_line)
csv1.writerow(type1_series) # Might need to convert to plain list if Series
# doesn't iterate the values you need directly
elif record_type == "type2":
type2_series = process_type2_line(json_line)
csv2.writerow(type2_series)
elif record_type == "type3":
type3_series = process_type3_line(json_line)
csv3.writerow(type3_series)
如果它按原樣作業（沒有執行者），請以這種方式使用它。如果您在沒有執行程式的情況下頁面抖動，或者檔案足夠大以至于重復append的 s 對您造成嚴重傷害，那么這可能足以使其完全獨立作業。如果速度太慢，如果您正在做大量作業將每一行處理成輸出格式，則執行程式可能會提供一點好處（因為在大多數作業人員正在處理時，一兩個作業人員可以充分共享磁盤訪問以進行讀寫） ，但是如果每條線的處理作業量很低，那么少數工人（我會從兩三個開始）只會增加磁盤爭用（特別是如果您使用的是旋轉磁盤硬碟驅動器，而不是 SSD )，并且并行性要么無濟于事，要么會積極傷害。
您可能需要調整使用的確切 CSV 方言csv.writer（作為引數encoding傳遞給檔案的消費者（s）期望），但這是一般形式。encoding='utf-8'encoding='utf-16'open.csv
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/476128.html