Как несколькими потоками читать файлы на питоне? Как ускорить следующий код?

Question

[LakeForest]

В столбце text вместо текста - пути к файлам с текстом, я хочу все тексты переписать в csv, но файлов очень много. Код обрабатывается дольше суток и до сих пор только 70%. Пишет, что примерно еще 12 часов ждать. Очень хочется ускорить, но плохо представляю, что с этим можно сделать...

c = 0
for i, row in tqdm(df_new.iterrows(), total=df_new.shape[0]):
    if "texts" not in row.text:
        continue
    c += 1
    with open("../../" + row.text, "r") as f:
        text = f.readline()
        df_new.loc[i, "text"] = text.replace("\n", "")
        if c % 10000 == 0:
            df_new.to_csv("df_all.csv", index=False)
df_new.to_csv("df_all.csv", index=False)

Answer 1

[Vindicar]

Несколько потоков тут вряд ли помогут из-за особенностей питона.
Несколько процессов - возможно (модуль multiprocessing).
Но вообще для начала стоит убедиться, что затык именно по CPU, а не по производительности диска.

Comments

[LakeForest]

Несколько процессов - возможно (модуль multiprocessing).

А что мне тогда в процессах распараллелить при помощи multiprocessing?

[Vindicar]

LakeForest, ну я думаю вам нужно посмотреть в сторону пула потоков, чтобы распараллелить цикл.
Правда, нужно иметь ввиду, что может быть проблематично передать объект в другой процесс - это должно быть что-то, что пройдёт через pickle. В идеале - комбинация списков/словарей и примитивных типов данных.

Но я ещё раз подчеркну - вы уверены, что узкое место - именно неэффективное использование CPU?
Например, если вы записываете в csv только каждый 10000й файл, зачем вы вообще открываете и обрабатываете остальные 9999 файлов?

[Сергей Горностаев]

Помогут, GIL не мешает вводу-выводу.

[LakeForest]

Подскажите, пожалуйста, что делаю не так? Результат в строки csv не сохраняется...

import multiprocessing

c1 = 0
class Result():
    def __init__(self, df):
        self.df = df

    def update_result(self, df):
        self.df = df

def process(file,i, df1):
    with open(file, "r") as f:
        text = f.readline()
        text = text.replace("\n", "")
        df1.loc[i, "text"] = text

    return df1

p = multiprocessing.Pool()
index = df1.text.str.contains("texts/")
result = Result(df1)
for i, row in tqdm(df1[index].iterrows(), total=df1[index].shape[0]):
    c1 += 1
    p.apply_async(process, ["../../" + row.text, i, df1], 
                       callback=result.update_result)
    
    if c1 % 10000 == 0:
        print(result.df[index].shape[0]) #не изменяется!
        result.df.to_csv("temp.csv", index=False)
p.close()
p.join()

[LakeForest]

Например, если вы записываете в csv только каждый 10000й файл, зачем вы вообще открываете и обрабатываете остальные 9999 файлов?

Vindicar, я все 10000 считанных файлов сохраняю каждую 10000 итераций, чтобы не потерять и заново код не запускать, если ноутбук упадет.

[LakeForest]

На global df1 вообще никакой реакции... Почему ничего не перезаписывается?

import multiprocessing

def process(objs):
    global df1
    with open(objs[0], "r") as f:
        text = f.readline()
        text = text.replace("\n", "")
        df1.loc[objs[1], "text"] = text


p = multiprocessing.Pool()

p.apply_async(process, [["../../texts/" + row.text, i] for i, row in tqdm(df1[index].iterrows(), total=df1[index].shape[0])], 
                   )
    
p.close()
p.join() # Wait for all child processes to close.

[Vindicar]

LakeForest, Я подозреваю, дело в том, что у вас создаётся копия df1 в каждом процессе.

def process(i, file):
    with open(file, "r") as f:
        text = f.readline()
    return i, text.replace("\n", "")

# ВАЖНО! иначе можно налететь на рекурсию
if __name__ == '__main__':
    data = ( i, "../../"+row.text for i, row in tqdm(
        df1[index].iterrows(), 
        total=df1[index].shape[0]
    ))
    with Pool(processes=4) as pool:
        for i, result in pool.imap_unordered(process, data):
            df1.loc[i, "text"] = result
            if i % 10000 == 0:
                try:
                    df1.to_csv("temp.csv", index=False)
                except IOError:
                    print("Can't save temp file")
    df1.to_csv("temp.csv", index=False)

[LakeForest]

Vindicar, странно, я увеличил Pool процессов до 10. Но поощущениям скорость совсем не изменилась. Так же как и моим самым первым способом пишет еще 6 часов будет обрабатываться...

[Vindicar]

LakeForest,
1. Не имеет особого смысла увеличивать число процессов больше числа ядер на процессоре.
2. Вы узкое место-то нашли? Или забили? Может, это периодическое сохранение в csv всё портит, а может ещё что. Например, у вас получается такая большая структура данных, что питон начинает дико свопиться.

[LakeForest]

Vindicar, до сохранения в csv еще дойти надо - это долгий процесс. Узкое место - чтение из файла (там по 1 короткой строке из каждого файла надо прочитать).