如何使用Python-Multiprocessing来限制许多文件/数据范围？

Question

我对Python和编程相对较新，只将其用于分析仿真数据。 我有一个目录“ result_1/”，其中有150000多个CSV文件，其中包含我想将pandas-dataframe限制的模拟数据。为了探索ReadDir（）一次仅读取32K目录条目的问题，我准备了“ files.csv” - 列出目录中的所有文件。

（“ sim”，“ det”和“ run”是我从文件名中读取的信息，并将其作为串联插入到数据框中。为了更好地忽略，我将它们的定义从concat中取出。）

我的问题如下： 该程序需要太多时间来运行，我想使用多处理/线程来加快循环的速度，但是由于我以前从未使用过MP/MT，我什至不知道此处是否可以使用它。

预先感谢您，祝您有美好的一天！

import numpy as np                          
import pandas as pd                         
import os
import multiprocessing as mp

df = pd.DataFrame()
path = 'result_1/'
list = pd.read_csv('files.csv', encoding='utf_16_le', names=['f'])['f'].values.tolist()

for file in list:
    dftemp = pd.read_csv(r'{}'.format(os.path.join(path, file)), skiprows=8, names=['x', 'y', 'z', 'dos'], sep=',').drop(['y', 'z'], axis=1)
    sim = pd.Series(int(file.split('Nr')[1].split('_')[0]) * np.ones((300,), dtype=int))
    det = pd.Series(int(file.split('Nr')[0]) * np.ones((300,), dtype=int))
    run = pd.Series(int(file[-8:-4]) * np.ones((300,), dtype=int))
    dftemp = pd.concat([sim, det, run, dftemp], axis=1)
    df = pd.concat([df, dftemp], axis=0)

df.rename({0:'sim', 1:'det', 2:'run', 3:'x', 4:'dos'}, axis=1).to_csv(r'df.csv')

CSV文件看起来像这样：“ 193nr6_run_0038.csv”（fe）

#(8 lines of things I don't need.)
0, 0, 0, 4.621046656438921e-09
1, 0, 0, 4.600856584602298e-09
(... 300 lines of data [x, y, z, dose])

Answer 1

由于CPU和RAM限制，并行处理数据框可能很困难。我不知道您的硬件规格或数据范围的详细信息。但是，我将使用多处理来“解析/制作”数据框，然后将它们串联。以下是一个示例：

import numpy as np                          
import pandas as pd                         
import os
from multiprocessing import Pool


path = 'result_1/'
list_of_files = pd.read_csv('files.csv', encoding='utf_16_le', names=['f'])['f'].values.tolist()

#make a function to replace the for-loop:
def my_custom_func(file):
    dftemp = pd.read_csv(r'{}'.format(os.path.join(path, file)), skiprows=8, names=['x', 'y', 'z', 'dos'], sep=',').drop(['y', 'z'], axis=1)
    sim = pd.Series(int(file.split('Nr')[1].split('_')[0]) * np.ones((300,), dtype=int))
    det = pd.Series(int(file.split('Nr')[0]) * np.ones((300,), dtype=int))
    run = pd.Series(int(file[-8:-4]) * np.ones((300,), dtype=int))
    return pd.concat([sim, det, run, dftemp], axis=1)

#use multiprocessing to process multiple files at once
with Pool(8) as p: #8 processes simultaneously. Avoid using more processes than cores in your CPU
    dataframes = p.map(my_custom_func, list_of_files)

#Finally, concatenate them all
df = pd.concat(dataframes)

df.rename({0:'sim', 1:'det', 2:'run', 3:'x', 4:'dos'}, axis=1).to_csv(r'df.csv')

请查看 有关更多信息。