читать несколько файлов, используя многопроцессорность

Question

Мне нужно прочитать несколько очень больших текстовых файлов (более 100 Мб), обработать каждую строку с помощью регулярного выражения и сохранить данные в структуре. Моя структура наследуется от defaultdict, у нее есть метод read (self), который читает файл self.file_name.

Посмотрите на этот очень простой (но не реальный) пример, я не использую регулярное выражение, но разбиваю строки:

import multiprocessing
from collections import defaultdict

def SingleContainer():
    return list()

class Container(defaultdict):
    """
    this class store odd line in self["odd"] and even line in self["even"].
    It is stupid, but it's only an example. In the real case the class
    has additional methods that do computation on readen data.
    """
    def __init__(self,file_name):
        if type(file_name) != str:
            raise AttributeError, "%s is not a string" % file_name
        defaultdict.__init__(self,SingleContainer)
        self.file_name = file_name
        self.readen_lines = 0
    def read(self):
        f = open(self.file_name)
        print "start reading file %s" % self.file_name
        for line in f:
            self.readen_lines += 1
            values = line.split()
            key = {0: "even", 1: "odd"}[self.readen_lines %2]
            self[key].append(values)
        print "readen %d lines from file %s" % (self.readen_lines, self.file_name)

def do(file_name):
    container = Container(file_name)
    container.read()
    return container.items()

if __name__ == "__main__":
    file_names = ["r1_200909.log", "r1_200910.log"]
    pool = multiprocessing.Pool(len(file_names))
    result = pool.map(do,file_names)
    pool.close()
    pool.join()
    print "Finish"      

В конце мне нужно объединить все результаты в один контейнер. Важно, что порядок линий сохраняется. Мой подход слишком медленный при возврате значений. Лучшее решение? Я использую Python 2.6 в Linux

Answer 1

Возможно, вы столкнулись с двумя проблемами.

Один из них был упомянут: вы читаете несколько файлов одновременно. Эти чтения в конечном итоге будут чередоваться, вызывая перебивание диска. Вы хотите прочитать целые файлы сразу, а затем только многопоточность вычислений на данных.

Во-вторых, у вас перегружены многопроцессорные модули Python. Это на самом деле не использование потоков, а запуск нескольких процессов и сериализация результатов по конвейеру. Это очень медленно для больших объемов данных - фактически, это кажется медленнее, чем работа, которую вы делаете в потоке (по крайней мере, в примере). Это реальная проблема, вызванная GIL.

Если я изменю do (), чтобы вернуть None вместо container.items (), чтобы отключить дополнительную копию данных, этот пример на быстрее, чем отдельный поток, если файлы уже кэшированы

Два потока: 0,36 прошло 168% ЦП

Один поток (замените файл pool.map на карту): 0: 00.52 прошло 98% ЦП

К сожалению, проблема GIL является фундаментальной и не может быть решена изнутри Python.

Answer 2

Вы создаете пул с таким количеством рабочих, сколько файлов.Это может быть слишком много.Обычно я стремлюсь, чтобы количество рабочих было примерно таким же, как и количество ядер.

Простой факт заключается в том, что ваш последний шаг будет одним процессом, объединяющим все результаты вместе.Этого не избежать, учитывая описание вашей проблемы.Это известно как барьерная синхронизация: все задачи должны достичь одной и той же точки, прежде чем любая из них может быть продолжена.

Вероятно, вам следует запускать эту программу несколько раз или в цикле, передавая различное значение multiprocessing.Pool() каждаявремя, начиная с 1 и до количества ядер.Определите время каждого прогона и посмотрите, какое количество рабочих мест работает лучше всего.

Результат будет зависеть от того, насколько сильно загружает процессор (в отличие от диска) ваша задача.Я не удивлюсь, если 2 будет лучше, если ваша задача - примерно половина процессора и половина диска, даже на 8-ядерном компьютере.

Answer 3

Многопроцессорная обработка больше подходит для процессов, ориентированных на процессор или память, поскольку время поиска вращающихся дисков снижает производительность при переключении между файлами. Либо загрузите файлы журналов на быструю флешку, либо на какой-нибудь диск памяти (физический или виртуальный), либо откажитесь от многопроцессорной обработки.