我正在處理大型 CSV 文件，我需要制作笛卡爾積（合并操作）。由于內存錯誤，我試圖解決 Pandas 的問題（您可以在此處查看 Pandas 的代碼和相同問題的數據格式示例）但沒有成功?，F在，我正在嘗試使用 Dask，它應該可以管理大量數據集，即使它的大小大于可用 RAM。首先，我閱讀了兩個 CSV：from dask import dataframe as ddBLOCKSIZE = 64000000 # = 64 Mb chunksdf1_file_path = './mRNA_TCGA_breast.csv'df2_file_path = './miRNA_TCGA_breast.csv'# Gets Dataframesdf1 = dd.read_csv( df1_file_path, delimiter='\t', blocksize=BLOCKSIZE)first_column = df1.columns.values[0]df1.set_index(first_column)df2 = dd.read_csv( df2_file_path, delimiter='\t', blocksize=BLOCKSIZE)first_column = df2.columns.values[0]df2.set_index(first_column)# Filter common columnscommon_columns = df1.columns.intersection(df2.columns)df1 = df1[common_columns]df2 = df2[common_columns]然后，我將操作存儲在磁盤上以防止內存錯誤：# Computes a Cartesian productdf1['_tmpkey'] = 1df2['_tmpkey'] = 1# Neither of these two options work# df1.merge(df2, on='_tmpkey').drop('_tmpkey', axis=1).to_hdf('/tmp/merge.*.hdf', key='/merge_data')# df1.merge(df2, on='_tmpkey').drop('_tmpkey', axis=1).to_parquet('/tmp/')我創建了一個 repo 來嘗試使用與我使用的完全相同的 CSV 文件。我嘗試過使用較小blocksize的值，但我得到了同樣的錯誤。我錯過了什么嗎？任何形式的幫助將非常感激。
1 回答
明月笑刀無情
TA貢獻1828條經驗 獲得超4個贊
我使用以下方法成功運行了您的代碼，內存限制為 32GB。
我已經擺脫了這個論點BLOCKSIZE，repartition而是在 df1 和 df2 上使用。
df1 = df1.repartition(npartitions=50)
df2 = df2.repartition(npartitions=1)
請注意，與 df1 相比，df2 的大小確實更?。?2.5 MB vs 23.75 MB），這就是為什么我只為 df2 保留一個分區并將 df1 分成 50 個分區的原因。
這樣做應該使代碼為您工作。對我來說，使用的內存保持在 12GB 以下。
為了檢查，我計算了結果的 len ：
len(df) # 3001995
按照上述內容創建一個包含 50 個分區的 parquet 文件。您可以repartition再次使用來獲得所需的 partition_size。
注意：
添加這個應該可以加快你的代碼：
from dask.distributed import Client
client = Client()
Client(processes=False)就我而言，由于我的運行環境，我不得不使用該參數。
添加回答
舉報
0/150
提交
取消