我有一個y.csv檔案。檔案大小為 10 MB，其中包含來自Jan 2020 to May 2020.
我每個月也有一個單獨的檔案。例如data-2020-01.csv。它包含詳細的資料。每個月檔案的檔案大小約為1 GB.
我y.csv按月拆分，然后通過加載相關的月份檔案來處理資料。當我花了很多個月的時間時，這個程序花費的時間太長了。例如 24 個月。
我想更快地處理資料。我可以訪問m6i.8xlarge具有記憶體32 vCPU的AWS 實體。128 GB
我是多處理的新手。那么有人可以在這里指導我嗎？
這是我當前的代碼。
import pandas as pd
periods = [(2020, 1), (2020, 2), (2020, 3), (2020, 4), (2020, 5)]
y = pd.read_csv("y.csv", index_col=0, parse_dates=True).fillna(0) # Filesize: ~10 MB
def process(_month_df, _index):
idx = _month_df.index[_month_df.index.get_loc(_index, method='nearest')]
for _, value in _month_df.loc[idx:].itertuples():
up_delta = 200
down_delta = 200
up_value = value up_delta
down_value = value - down_delta
if value > up_value:
y.loc[_index, "result"] = 1
return
if value < down_value:
y.loc[_index, "result"] = 0
return
for x in periods:
filename = "data-" str(x[0]) "-" str(x[1]).zfill(2) # data-2020-01
filtered_y = y[(y.index.month == x[1]) & (y.index.year == x[0])] # Only get the current month records
month_df = pd.read_csv(f'{filename}.csv', index_col=0, parse_dates=True) # Filesize: ~1 GB (data-2020-01.csv)
for index, row in filtered_y.iterrows():
process(month_df, index)
uj5u.com熱心網友回復：
多執行緒池非常適合y在執行緒之間共享資料幀（無需使用共享記憶體），但不太擅長并行運行 CPU 密集型處理。多處理池非常適合進行 CPU 密集型處理，但在跨行程共享資料方面不太好，而沒有提出資料幀的碎片記憶體表示y。
在這里，我重新安排了您的代碼，以便我使用多執行緒池filtered_y為每個周期創建（這是一個 CPU 密集型操作，但 pandas 確實為某些操作釋放了全域解釋器鎖——希望是這個）。然后我們只是將一個月的資料傳遞給一個多處理池，而不是整個y資料幀，以使用 worker function 處理那個月process_month。但由于每個池行程無權訪問y資料幀，它只回傳需要用要替換的值更新的索引。
import pandas as pd
from multiprocessing.pool import Pool, ThreadPool
def process_month(period, filtered_y):
"""
returns a list of tuples consisting of (index, value) pairs
"""
filename = "data-" str(period[0]) "-" str(period[1]).zfill(2) # data-2020-01
month_df = pd.read_csv(f'{filename}.csv', index_col=0, parse_dates=True) # Filesize: ~1 GB (data-2020-01.csv)
results = []
for index, row in filtered_y.iterrows():
idx = month_df.index[month_df.index.get_loc(index, method='nearest')]
for _, value in month_df.loc[idx:].itertuples():
up_delta = 200
down_delta = 200
up_value = value up_delta
down_value = value - down_delta
if value > up_value:
results.append((index, 1))
break
if value < down_value:
results.append((index, 0))
break
return results
def process(period):
filtered_y = y[(y.index.month == period[1]) & (y.index.year == period[0])] # Only get the current month records
for index, value in multiprocessing_pool.apply(process_month(period, filtered_y)):
y.loc[index, "result"] = value
def main():
global y, multiprocessing_pool
periods = [(2020, 1), (2020, 2), (2020, 3), (2020, 4), (2020, 5)]
y = pd.read_csv("y.csv", index_col=0, parse_dates=True).fillna(0) # Filesize: ~10 MB
with Pool() as multiprocessing_pool, ThreadPool(len(periods)) as thread_pool:
thread_pool.map(process, periods)
# Presumably y gets written out again as a CSV file here?
# Required for Windows:
if __name__ == '__main__':
main()
uj5u.com熱心網友回復：
正如在多個熊貓/執行緒問題中所評論的，CSV 檔案是 IO 系結的，您可以從使用ThreadPoolExecutor中獲得一些好處。
同時，如果您要執行聚合操作，請考慮在您的處理器內部read_csv執行，并改用ProcessPoolExecutor。
如果您要在多行程之間傳遞大量資料，您還需要適當的記憶體共享方法。
但是，我看到了 和 的使用，iterrows總的來說，itertuples這兩個指令讓我的眼睛流血了。您確定不能以矢量化模式處理資料嗎？
這個特定的部分我不確定它應該做什么，并且有 M 行會使其非常慢。
def process(_month_df, _index):
idx = _month_df.index[_month_df.index.get_loc(_index, method='nearest')]
for _, value in _month_df.loc[idx:].itertuples():
up_delta = 200
down_delta = 200
up_value = value up_delta
down_value = value - down_delta
if value > up_value:
y.loc[_index, "result"] = 1
return
if value < down_value:
y.loc[_index, "result"] = 0
return
在矢量化代碼下方查找它是上升還是下降，以及在哪一行
df=pd.DataFrame({'vals': np.random.random(int(10))*1000 5000}).astype('int64')
print(df.vals.values)
up_value = 6000
down_value = 3000
valsup = df.vals.values 200*np.arange(df.shape[0]) 200
valsdown = df.vals.values - 200*np.arange(df.shape[0])-200
#! argmax returns 0 if all false
# idx_up = np.argmax(valsup > up_value)
# idx_dwn= np.argmax(valsdown < down_value)
idx_up = np.argwhere(valsup > up_value)
idx_dwn= np.argwhere(valsdown < down_value)
idx_up = idx_up[0][0] if len(idx_up) else -1
idx_dwn = idx_dwn[0][0] if len(idx_dwn) else -1
if idx_up < 0 and idx_dwn<0:
print(f" Not up nor down")
if idx_up < idx_dwn or idx_dwn<0:
print(f" Result is positive, in position {idx_up}")
else:
print(f" Result is negative, in position {idx_dwn}")
為了完整性，基準測驗itertuples()和argwhere1000 個元素的方法：
.itertuples(): 757μsarangeargwhere: 60μs
轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/479711.html
標籤：Python python-3.x 熊猫 多处理
上一篇：關于Pandas資料框