Как решить проблемы с памятью при многопроцессорной обработке с помощью Pool.map()?

Я написал программу (ниже) для:

прочитайте огромный текстовый файл как pandas dataframe
затем groupby используя определенное значение столбца, чтобы разделить данные и сохранить в виде списка данных.
затем multiprocess Pool.map() данные в multiprocess Pool.map() для параллельной обработки каждого кадра данных.

Все хорошо, программа хорошо работает с моим небольшим набором тестовых данных. Но когда я передаю большие данные (около 14 ГБ), потребление памяти экспоненциально увеличивается, а затем зависает или убивается (в кластере HPC).

Я добавил коды, чтобы очистить память, как только данные/переменная бесполезны. Я также закрываю бассейн, как только это будет сделано. Тем не менее, при входе в 14 ГБ я ожидал только 2 * 14 ГБ памяти, но, похоже, многое происходит. Я также пытался настроить, используя chunkSize and maxTaskPerChild, etc Но я не вижу никакой разницы в оптимизации в обоих тестах и больших файлах.

Я думаю, что улучшения в этом коде необходимы в этой позиции кода, когда я начинаю multiprocessing.

p = Pool(3) # number of pool to run at once; default at 1 result = p.map(matrix_to_vcf, list(gen_matrix_df_list.values())) p = Pool(3) # number of pool to run at once; default at 1 result = p.map(matrix_to_vcf, list(gen_matrix_df_list.values())) но я p = Pool(3) # number of pool to run at once; default at 1 result = p.map(matrix_to_vcf, list(gen_matrix_df_list.values())) весь код.

Тестовый пример: я создал тестовый файл ("genome_matrix_final-chr1234-1mb.txt") объемом до 250 МБ и запустил программу. Когда я проверяю системный монитор, я вижу, что потребление памяти увеличилось примерно на 6 ГБ. Мне не очень понятно, почему так много места в памяти занимает файл 250 Мб плюс некоторые выводы. Я поделился этим файлом через выпадающий список, если он помогает увидеть реальную проблему. https://www.dropbox.com/sh/coihujii38t5prd/AABDXv8ACGIYczeMtzKBo0eea?dl=0

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

Мой скрипт на python:

#!/home/bin/python3

import pandas as pd
import collections
from multiprocessing import Pool
import io
import time
import resource

print()
print('Checking required modules')
print()


''' change this input file name and/or path as need be '''
genome_matrix_file = "genome_matrix_final-chr1n2-2mb.txt"   # test file 01
genome_matrix_file = "genome_matrix_final-chr1234-1mb.txt"  # test file 02
#genome_matrix_file = "genome_matrix_final.txt"    # large file 

def main():
    with open("genome_matrix_header.txt") as header:
        header = header.read().rstrip('\n').split('\t')
        print()

    time01 = time.time()
    print('starting time: ', time01)

    '''load the genome matrix file onto pandas as dataframe.
    This makes is more easy for multiprocessing'''
    gen_matrix_df = pd.read_csv(genome_matrix_file, sep='\t', names=header)

    # now, group the dataframe by chromosome/contig - so it can be multiprocessed
    gen_matrix_df = gen_matrix_df.groupby('CHROM')

    # store the splitted dataframes as list of key, values(pandas dataframe) pairs
    # this list of dataframe will be used while multiprocessing
    gen_matrix_df_list = collections.OrderedDict()
    for chr_, data in gen_matrix_df:
        gen_matrix_df_list[chr_] = data

    # clear memory
    del gen_matrix_df

    '''Now, pipe each dataframe from the list using map.Pool() '''
    p = Pool(3)  # number of pool to run at once; default at 1
    result = p.map(matrix_to_vcf, list(gen_matrix_df_list.values()))

    del gen_matrix_df_list  # clear memory

    p.close()
    p.join()


    # concat the results from pool.map() and write it to a file
    result_merged = pd.concat(result)
    del result  # clear memory

    pd.DataFrame.to_csv(result_merged, "matrix_to_haplotype-chr1n2.txt", sep='\t', header=True, index=False)

    print()
    print('completed all process in "%s" sec. ' % (time.time() - time01))
    print('Global maximum memory usage: %.2f (mb)' % current_mem_usage())
    print()


'''function to convert the dataframe from genome matrix to desired output '''
def matrix_to_vcf(matrix_df):

    print()
    time02 = time.time()

    # index position of the samples in genome matrix file
    sample_idx = [{'10a': 33, '10b': 18}, {'13a': 3, '13b': 19},
                    {'14a': 20, '14b': 4}, {'16a': 5, '16b': 21},
                    {'17a': 6, '17b': 22}, {'23a': 7, '23b': 23},
                    {'24a': 8, '24b': 24}, {'25a': 25, '25b': 9},
                    {'26a': 10, '26b': 26}, {'34a': 11, '34b': 27},
                    {'35a': 12, '35b': 28}, {'37a': 13, '37b': 29},
                    {'38a': 14, '38b': 30}, {'3a': 31, '3b': 15},
                    {'8a': 32, '8b': 17}]

    # sample index stored as ordered dictionary
    sample_idx_ord_list = []
    for ids in sample_idx:
        ids = collections.OrderedDict(sorted(ids.items()))
        sample_idx_ord_list.append(ids)


    # for haplotype file
    header = ['contig', 'pos', 'ref', 'alt']

    # adding some suffixes "PI" to available sample names
    for item in sample_idx_ord_list:
        ks_update = ''
        for ks in item.keys():
            ks_update += ks
        header.append(ks_update+'_PI')
        header.append(ks_update+'_PG_al')


    #final variable store the haplotype data
    # write the header lines first
    haplotype_output = '\t'.join(header) + '\n'


    # to store the value of parsed the line and update the "PI", "PG" value for each sample
    updated_line = ''

    # read the piped in data back to text like file
    matrix_df = pd.DataFrame.to_csv(matrix_df, sep='\t', index=False)

    matrix_df = matrix_df.rstrip('\n').split('\n')
    for line in matrix_df:
        if line.startswith('CHROM'):
            continue

        line_split = line.split('\t')
        chr_ = line_split[0]
        ref = line_split[2]
        alt = list(set(line_split[3:]))

        # remove the alleles "N" missing and "ref" from the alt-alleles
        alt_up = list(filter(lambda x: x!='N' and x!=ref, alt))

        # if no alt alleles are found, just continue
        # - i.e : don't write that line in output file
        if len(alt_up) == 0:
            continue

        #print('\nMining data for chromosome/contig "%s" ' %(chr_ ))
        #so, we have data for CHR, POS, REF, ALT so far
        # now, we mine phased genotype for each sample pair (as "PG_al", and also add "PI" tag)
        sample_data_for_vcf = []
        for ids in sample_idx_ord_list:
            sample_data = []
            for key, val in ids.items():
                sample_value = line_split[val]
                sample_data.append(sample_value)

            # now, update the phased state for each sample
            # also replacing the missing allele i.e "N" and "-" with ref-allele
            sample_data = ('|'.join(sample_data)).replace('N', ref).replace('-', ref)
            sample_data_for_vcf.append(str(chr_))
            sample_data_for_vcf.append(sample_data)

        # add data for all the samples in that line, append it with former columns (chrom, pos ..) ..
        # and .. write it to final haplotype file
        sample_data_for_vcf = '\t'.join(sample_data_for_vcf)
        updated_line = '\t'.join(line_split[0:3]) + '\t' + ','.join(alt_up) + \
            '\t' + sample_data_for_vcf + '\n'
        haplotype_output += updated_line

    del matrix_df  # clear memory
    print('completed haplotype preparation for chromosome/contig "%s" '
          'in "%s" sec. ' %(chr_, time.time()-time02))
    print('\tWorker maximum memory usage: %.2f (mb)' %(current_mem_usage()))

    # return the data back to the pool
    return pd.read_csv(io.StringIO(haplotype_output), sep='\t')


''' to monitor memory '''
def current_mem_usage():
    return resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024.


if __name__ == '__main__':
    main()

Обновление для охотников за головами:

Я достиг многопроцессорной обработки с помощью Pool.map() но код вызывает большой объем памяти (входной тестовый файл ~ 300 МБ, но объем памяти составляет около 6 ГБ). Я ожидал только 3 * 300 МБ памяти на макс.

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

Ответы

Ответ 1

необходимое условие

В Python (далее я использую 64-битную сборку Python 3.6.5) все является объектом. Это имеет свои накладные расходы, и с помощью getsizeof мы можем видеть точно размер объекта в байтах:
```
>>> import sys
>>> sys.getsizeof(42)
28
>>> sys.getsizeof('T')
50
```
Когда для создания дочернего процесса используется системный вызов fork (по умолчанию * nix, см. multiprocessing.get_start_method()), родительская физическая память не копируется и используется метод копирования -o n-write.
Дочерний процесс Fork по-прежнему будет сообщать полный RSS (размер резидентного набора) родительского процесса. В связи с этим PSS (пропорциональный размер набора) является более подходящим показателем для оценки использования памяти приложения разветвления. Вот пример со страницы:

Процесс А имеет 50 КиБ неразделенной памяти

Процесс B имеет 300 КБ неразделенной памяти

И процесс A, и процесс B имеют 100 КиБ одной и той же области общей памяти
Поскольку PSS определяется как сумма неразделенной памяти процесса и доли памяти, используемой совместно с другими процессами, PSS для этих двух процессов выглядит следующим образом:

PSS процесса A = 50 КиБ + (100 КиБ /2) = 100 КиБ

PSS процесса B = 300 КиБ + (100 КиБ /2) = 350 КиБ

Фрейм данных

Не давайте смотреть на ваш DataFrame одиночку. memory_profiler поможет нам.

justpd.py

#!/usr/bin/env python3

import pandas as pd
from memory_profiler import profile

@profile
def main():
    with open('genome_matrix_header.txt') as header:
        header = header.read().rstrip('\n').split('\t')

    gen_matrix_df = pd.read_csv(
        'genome_matrix_final-chr1234-1mb.txt', sep='\t', names=header)

    gen_matrix_df.info()
    gen_matrix_df.info(memory_usage='deep')

if __name__ == '__main__':
    main()

Теперь давайте использовать профилировщик:

mprof run justpd.py
mprof plot

Мы можем увидеть сюжет:

и построчная -l In trace:

Line #    Mem usage    Increment   Line Contents
================================================
     6     54.3 MiB     54.3 MiB   @profile
     7                             def main():
     8     54.3 MiB      0.0 MiB       with open('genome_matrix_header.txt') as header:
     9     54.3 MiB      0.0 MiB           header = header.read().rstrip('\n').split('\t')
    10                             
    11   2072.0 MiB   2017.7 MiB       gen_matrix_df = pd.read_csv('genome_matrix_final-chr1234-1mb.txt', sep='\t', names=header)
    12                                 
    13   2072.0 MiB      0.0 MiB       gen_matrix_df.info()
    14   2072.0 MiB      0.0 MiB       gen_matrix_df.info(memory_usage='deep')

Мы можем видеть, что кадр данных занимает ~ 2 ГиБ с пиком в ~ 3 ГиБ во время его построения. Что более интересно, так это вывод info.

<class 'pandas.core.frame.DataFrame'>
RangeIndex: 4000000 entries, 0 to 3999999
Data columns (total 34 columns):
...
dtypes: int64(2), object(32)
memory usage: 1.0+ GB

Но info(memory_usage='deep') ("deep" означает глубокий самоанализ данных путем опроса object dtype s, см. Ниже) дает:

memory usage: 7.9 GB

Да?! Посмотрев за пределы процесса, мы можем убедиться, что цифры memory_profiler верны. sys.getsizeof также показывает то же значение для фрейма (наиболее вероятно из-за пользовательского __sizeof__), как и другие инструменты, которые используют его для оценки выделенного gc.get_objects(), например, pympler.

# added after read_csv
from pympler import tracker
tr = tracker.SummaryTracker()
tr.print_diff()

дает:

                                             types |   # objects |   total size
================================================== | =========== | ============
                 <class 'pandas.core.series.Series |          34 |      7.93 GB
                                      <class 'list |        7839 |    732.38 KB
                                       <class 'str |        7741 |    550.10 KB
                                       <class 'int |        1810 |     49.66 KB
                                      <class 'dict |          38 |      7.43 KB
  <class 'pandas.core.internals.SingleBlockManager |          34 |      3.98 KB
                             <class 'numpy.ndarray |          34 |      3.19 KB

Так откуда берутся эти 7,93 ГиБ? Давайте попробуем объяснить это. У нас есть 4M строк и 34 столбца, что дает нам 134M значений. Это либо int64 либо object (который является 64-битным указателем; подробное объяснение см. В разделе " Использование панд с большими данными"). Таким образом, у нас 134 * 10 ** 6 * 8/2 ** 20 ~ 1022 МБ только для значений в кадре данных. Как насчет оставшихся ~ 6,93 ГиБ?

Струнная интернирование

Чтобы понять поведение, необходимо знать, что Python выполняет интернирование строк. Есть две хорошие статьи (одна, две) об интернировании строк в Python 2. Помимо изменения Unicode в Python 3 и PEP 393 в Python 3.3, C-структуры изменились, но идея та же. По сути, каждая короткая строка, которая выглядит как идентификатор, будет кэшироваться Python во внутреннем словаре, а ссылки будут указывать на одни и те же объекты Python. Другими словами, мы можем сказать, что он ведет себя как одиночка. Статьи, которые я упоминал выше, объясняют, какие значительные улучшения в профиле памяти и производительности он дает. Мы можем проверить, если строка интернированы с использованием interned поле PyASCIIObject:

import ctypes

class PyASCIIObject(ctypes.Structure):
     _fields_ = [
         ('ob_refcnt', ctypes.c_size_t),
         ('ob_type', ctypes.py_object),
         ('length', ctypes.c_ssize_t),
         ('hash', ctypes.c_int64),
         ('state', ctypes.c_int32),
         ('wstr', ctypes.c_wchar_p)
    ]

Затем:

>>> a = 'name'
>>> b = '[email protected]#$'
>>> a_struct = PyASCIIObject.from_address(id(a))
>>> a_struct.state & 0b11
1
>>> b_struct = PyASCIIObject.from_address(id(b))
>>> b_struct.state & 0b11
0

С помощью двух строк мы также можем сравнивать идентификаторы (в случае CPython это делается при сравнении памяти).

>>> a = 'foo'
>>> b = 'foo'
>>> a is b
True
>> gen_matrix_df.REF[0] is gen_matrix_df.REF[6]
True

Из-за этого факта в отношении object dtype фрейм данных выделяет не более 20 строк (по одной на аминокислоты). Однако стоит отметить, что Pandas рекомендует категориальные типы для перечислений.

Память панд

Таким образом, мы можем объяснить наивную оценку в 7,93 ГиБ как:

>>> rows = 4 * 10 ** 6
>>> int_cols = 2
>>> str_cols = 32
>>> int_size = 8
>>> str_size = 58  
>>> ptr_size = 8
>>> (int_cols * int_size + str_cols * (str_size + ptr_size)) * rows / 2 ** 30
7.927417755126953

Обратите внимание, что str_size составляет 58 байт, а не 50, как мы видели выше для 1-символьного литерала. Это потому, что PEP 393 определяет компактные и некомпактные строки. Вы можете проверить это с помощью sys.getsizeof(gen_matrix_df.REF[0]).

Фактическое потребление памяти должно составлять ~ 1 ГиБ, как сообщает gen_matrix_df.info(), это в два раза больше. Можно предположить, что это как-то связано с (предварительным) распределением памяти, выполняемым Pandas или NumPy. Следующий эксперимент показывает, что это не без причины (несколько прогонов показывают картинку сохранения):

Line #    Mem usage    Increment   Line Contents
================================================
     8     53.1 MiB     53.1 MiB   @profile
     9                             def main():
    10     53.1 MiB      0.0 MiB       with open("genome_matrix_header.txt") as header:
    11     53.1 MiB      0.0 MiB           header = header.read().rstrip('\n').split('\t')
    12                             
    13   2070.9 MiB   2017.8 MiB       gen_matrix_df = pd.read_csv('genome_matrix_final-chr1234-1mb.txt', sep='\t', names=header)
    14   2071.2 MiB      0.4 MiB       gen_matrix_df = gen_matrix_df.drop(columns=[gen_matrix_df.keys()[0]])
    15   2071.2 MiB      0.0 MiB       gen_matrix_df = gen_matrix_df.drop(columns=[gen_matrix_df.keys()[0]])
    16   2040.7 MiB    -30.5 MiB       gen_matrix_df = gen_matrix_df.drop(columns=[random.choice(gen_matrix_df.keys())])
    ...
    23   1827.1 MiB    -30.5 MiB       gen_matrix_df = gen_matrix_df.drop(columns=[random.choice(gen_matrix_df.keys())])
    24   1094.7 MiB   -732.4 MiB       gen_matrix_df = gen_matrix_df.drop(columns=[random.choice(gen_matrix_df.keys())])
    25   1765.9 MiB    671.3 MiB       gen_matrix_df = gen_matrix_df.drop(columns=[random.choice(gen_matrix_df.keys())])
    26   1094.7 MiB   -671.3 MiB       gen_matrix_df = gen_matrix_df.drop(columns=[random.choice(gen_matrix_df.keys())])
    27   1704.8 MiB    610.2 MiB       gen_matrix_df = gen_matrix_df.drop(columns=[random.choice(gen_matrix_df.keys())])
    28   1094.7 MiB   -610.2 MiB       gen_matrix_df = gen_matrix_df.drop(columns=[random.choice(gen_matrix_df.keys())])
    29   1643.9 MiB    549.2 MiB       gen_matrix_df = gen_matrix_df.drop(columns=[random.choice(gen_matrix_df.keys())])
    30   1094.7 MiB   -549.2 MiB       gen_matrix_df = gen_matrix_df.drop(columns=[random.choice(gen_matrix_df.keys())])
    31   1582.8 MiB    488.1 MiB       gen_matrix_df = gen_matrix_df.drop(columns=[random.choice(gen_matrix_df.keys())])
    32   1094.7 MiB   -488.1 MiB       gen_matrix_df = gen_matrix_df.drop(columns=[random.choice(gen_matrix_df.keys())])    
    33   1521.9 MiB    427.2 MiB       gen_matrix_df = gen_matrix_df.drop(columns=[random.choice(gen_matrix_df.keys())])    
    34   1094.7 MiB   -427.2 MiB       gen_matrix_df = gen_matrix_df.drop(columns=[random.choice(gen_matrix_df.keys())])
    35   1460.8 MiB    366.1 MiB       gen_matrix_df = gen_matrix_df.drop(columns=[random.choice(gen_matrix_df.keys())])
    36   1094.7 MiB   -366.1 MiB       gen_matrix_df = gen_matrix_df.drop(columns=[random.choice(gen_matrix_df.keys())])
    37   1094.7 MiB      0.0 MiB       gen_matrix_df = gen_matrix_df.drop(columns=[random.choice(gen_matrix_df.keys())])
    ...
    47   1094.7 MiB      0.0 MiB       gen_matrix_df = gen_matrix_df.drop(columns=[random.choice(gen_matrix_df.keys())])

Я хочу закончить этот раздел цитатой из свежей статьи о проблемах дизайна и будущих Pandas2 от первоначального автора Pandas.

Практическое правило панд: объем оперативной памяти в 5-10 раз больше размера вашего набора данных

Дерево процессов

Наконец, давайте подойдем к пулу и посмотрим, сможет ли использовать copy -o n-write. Мы будем использовать smemstat (доступный из репозитория Ubuntu), чтобы оценить совместное использование памяти группой процессов и glances для записи свободной памяти всей системы. Оба могут написать JSON.

Запустим оригинальный скрипт с Pool(2). Нам понадобятся 3 оконных окна.

smemstat -l -M -p "python3.6 script.py" -o smemstat.json 1
glances -t 1 --export-json glances.json
mprof run -M script.py

Тогда mprof plot производит:

Диаграмма сумм (mprof run --nopython --include-children./script.py) выглядит следующим образом:

Обратите внимание, что две диаграммы выше показывают RSS. Гипотеза состоит в том, что из-за копирования -o n-write это не отражает фактическое использование памяти. Теперь у нас есть два JSON файлов из smemstat и glances. Я приведу следующий скрипт для преобразования файлов JSON в CSV.

#!/usr/bin/env python3

import csv
import sys
import json

def smemstat():
  with open('smemstat.json') as f:
    smem = json.load(f)

  rows = []
  fieldnames = set()    
  for s in smem['smemstat']['periodic-samples']:
    row = {}
    for ps in s['smem-per-process']:
      if 'script.py' in ps['command']:
        for k in ('uss', 'pss', 'rss'):
          row['{}-{}'.format(ps['pid'], k)] = ps[k] // 2 ** 20

    # smemstat produces empty samples, backfill from previous
    if rows:            
      for k, v in rows[-1].items():
        row.setdefault(k, v)

    rows.append(row)
    fieldnames.update(row.keys())

  with open('smemstat.csv', 'w') as out:
    dw = csv.DictWriter(out, fieldnames=sorted(fieldnames))
    dw.writeheader()
    list(map(dw.writerow, rows))

def glances():
  rows = []
  fieldnames = ['available', 'used', 'cached', 'mem_careful', 'percent',
    'free', 'mem_critical', 'inactive', 'shared', 'history_size',
    'mem_warning', 'total', 'active', 'buffers']
  with open('glances.csv', 'w') as out:
    dw = csv.DictWriter(out, fieldnames=fieldnames)
    dw.writeheader()
    with open('glances.json') as f:
      for l in f:
        d = json.loads(l)
        dw.writerow(d['mem'])

if __name__ == '__main__':
  globals()[sys.argv[1]]()

Сначала давайте посмотрим на free память.

Разница между первым и минимальным составляет ~ 4,15 ГиБ. А вот как выглядят цифры PSS:

И сумма:

Таким образом, мы видим, что из-за копирования -o n-запись фактическое потребление памяти составляет ~ 4,15 ГБ. Но мы все еще сериализуем данные для отправки их рабочим процессам через Pool.map. Можем ли мы использовать здесь копию -o n-write?

Общие данные

Чтобы использовать copy -o n-write, нам нужно, чтобы list(gen_matrix_df_list.values()) был доступен глобально, чтобы рабочий после fork все еще мог его прочитать.

Позвольте изменить код после del gen_matrix_df в main следующим образом:

...
global global_gen_matrix_df_values
global_gen_matrix_df_values = list(gen_matrix_df_list.values())
del gen_matrix_df_list

p = Pool(2)
result = p.map(matrix_to_vcf, range(len(global_gen_matrix_df_values)))
...

Удалить del gen_matrix_df_list который идет позже.
И измените первые строки matrix_to_vcf следующим образом:
```
def matrix_to_vcf(i):
    matrix_df = global_gen_matrix_df_values[i]
```

Теперь давайте перезапустим его. Свободная память:

Дерево процессов:

И его сумма:

Таким образом, мы не превышаем ~ 2,9 ГБ фактического использования памяти (пик основного процесса имеет место при построении фрейма данных), и копирование -o n-write помогло!

Как примечание, есть так называемое copy -o n-read, поведение сборщика мусора ссылочного цикла Python, описанное в Instagram Engineering (которое привело к gc.freeze в выпуске 315858). Но gc.disable() не оказывает влияния в данном конкретном случае.

Обновить

Альтернативой копированию -o n-write copy -l ess данных может быть делегирование его ядру с самого начала с помощью numpy.memmap. Вот пример реализации из High Performance Data Processing в Python. Хитрость затем сделать Панды использовать mmaped массив Numpy.

Ответ 2

Я была такая же проблема. Мне нужно было обработать огромный текстовый корпус, сохранив базу знаний из нескольких DataFrames из миллионов строк, загруженных в память. Я думаю, что этот вопрос распространен, поэтому я буду держать свой ответ ориентированным на общие цели.

Комбинация настроек решила проблему для меня (1 и 3 и 5 могут сделать это только для вас):

Используйте Pool.imap (или imap_unordered) вместо Pool.map. Это приведет к переходу данных по лени, чем загрузка всего его в память перед началом обработки.
Задайте значение для параметра chunksize. Это также сделает imap быстрее.
Задайте значение для параметра maxtasksperchild.
Добавить вывод на диск, чем в памяти. Мгновенно или каждый раз, когда он достигает определенного размера.
Запустите код в разных партиях. Вы можете использовать itertools.islice, если у вас есть итератор. Идея состоит в том, чтобы разделить ваш list(gen_matrix_df_list.values()) на три или более списка, затем вы передаете первую треть только для map или imap, затем второй трети в другой перспективе и т.д. Поскольку у вас есть список, вы можете просто отрежьте его в той же строке кода.

Ответ 3

Когда вы используете multiprocessing.Pool создан ряд дочерних процессов с использованием системного вызова fork(). Каждый из этих процессов начинается с точной копии памяти родительского процесса в то время. Поскольку вы загружаете CSV, прежде чем создавать Pool размером 3, каждый из этих трех процессов в пуле будет излишне иметь копию фрейма данных. (gen_matrix_df а также gen_matrix_df_list будет существовать как в текущем процессе, так и в каждом из 3 дочерних процессов, поэтому в памяти будет 4 копии каждой из этих структур)

Попробуйте создать Pool перед загрузкой файла (в самом начале на самом деле). Это должно уменьшить использование памяти.

Если он все еще слишком высок, вы можете:

Дамп gen_matrix_df_list в файл, по 1 позиции на строку, например:

import os
import cPickle

with open('tempfile.txt', 'w') as f:
    for item in gen_matrix_df_list.items():
        cPickle.dump(item, f)
        f.write(os.linesep)

Используйте Pool.imap() на итераторе по строкам, которые вы сбросили в этот файл, например:
```
with open('tempfile.txt', 'r') as f:
    p.imap(matrix_to_vcf, (cPickle.loads(line) for line in f))
```
(Обратите внимание, что matrix_to_vcf берет matrix_to_vcf (key, value) в приведенном выше примере, а не только значение)

Надеюсь, это поможет.

NB: Я не тестировал код выше. Это только предназначалось для демонстрации идеи.

Ответ 4

ОБЩИЙ ОТВЕТ О ПАМЯТИ С МНОГООБРАБОТКОЙ

Вы спросили: "Что вызывает выделение так много памяти". Ответ основывается на двух частях.

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

По умолчанию рабочие пула являются реальными процессами Python, раздвоенными с использованием модуля многопроцессорности стандартной библиотеки Python, когда n_jobs! = 1. Аргументы, переданные в качестве входа в вызов Parallel, сериализуются и перераспределяются в памяти каждого рабочего процесса.

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

Во-вторых, если вы пытаетесь восстановить память, вам нужно понять, что python работает иначе, чем другие языки, и вы полагаетесь на del, чтобы освободить память, когда этого не происходит. Я не знаю, лучше ли это, но в моем собственном коде я преодолел это, переопределив переменную в Пустое или пустое.

ДЛЯ ВАШЕГО СПЕЦИФИЧЕСКОГО ПРИМЕРА - РЕДАКТИРОВАНИЕ МИНИМАЛЬНОГО КОДА

До тех пор, пока вы можете поместить ваши большие данные в память дважды, я думаю, вы можете делать то, что вы пытаетесь сделать, просто изменив одну строку. Я написал очень похожий код, и это сработало для меня, когда я переназначил переменную (прокси-вызов del или какой-либо сбор мусора). Если это не сработает, вам может потребоваться следовать приведенным выше предложениям и использовать дисковый ввод-вывод:

    #### earlier code all the same
    # clear memory by reassignment (not del or gc)
    gen_matrix_df = {}

    '''Now, pipe each dataframe from the list using map.Pool() '''
    p = Pool(3)  # number of pool to run at once; default at 1
    result = p.map(matrix_to_vcf, list(gen_matrix_df_list.values()))

    #del gen_matrix_df_list  # I suspect you don't even need this, memory will free when the pool is closed

    p.close()
    p.join()
    #### later code all the same

ДЛЯ ВАШЕГО СПЕЦИФИЧЕСКОГО ПРИМЕРА - ОПТИМАЛЬНОЕ ИСПОЛЬЗОВАНИЕ ПАМЯТИ

До тех пор, пока вы можете поместить свои большие данные в память один раз, и у вас есть представление о том, насколько велик ваш файл, вы можете использовать чтение частичного файла Pandas read_csv, чтобы читать только в одно время, если вы действительно хотите управлять микроуправлением сколько данных считывается, или [фиксированный объем памяти за раз, используя chunksize], который возвращает итератор 5. Под этим я подразумеваю, что параметр nrows - это всего лишь одно чтение: вы можете использовать это, чтобы просто заглянуть в файл, или если по какой-то причине вы хотели, чтобы каждая часть имела точно такое же количество строк (потому что, например, если какая-либо из ваших данных представляет собой строки переменной длины, каждая строка не будет занимать одинаковый объем памяти). Но я думаю, что в целях подготовки файла для многопроцессорной обработки гораздо проще использовать куски, потому что это напрямую связано с памятью, что вас беспокоит. Будет проще использовать пробную версию и ошибку, чтобы вписаться в память на основе определенных размеров, чем количество строк, что изменит объем использования памяти в зависимости от того, сколько данных в строках. Единственная другая сложная часть заключается в том, что по какой-то конкретной причине приложения вы группируете несколько строк, поэтому это немного усложняет ситуацию. Используя ваш код в качестве примера:

   '''load the genome matrix file onto pandas as dataframe.
    This makes is more easy for multiprocessing'''

    # store the splitted dataframes as list of key, values(pandas dataframe) pairs
    # this list of dataframe will be used while multiprocessing
    #not sure why you need the ordered dict here, might add memory overhead
    #gen_matrix_df_list = collections.OrderedDict()  
    #a defaultdict won't throw an exception when we try to append to it the first time. if you don't want a default dict for some reason, you have to initialize each entry you care about.
    gen_matrix_df_list = collections.defaultdict(list)   
    chunksize = 10 ** 6

    for chunk in pd.read_csv(genome_matrix_file, sep='\t', names=header, chunksize=chunksize)
        # now, group the dataframe by chromosome/contig - so it can be multiprocessed
        gen_matrix_df = chunk.groupby('CHROM')
        for chr_, data in gen_matrix_df:
            gen_matrix_df_list[chr_].append(data)

    '''Having sorted chunks on read to a list of df, now create single data frames for each chr_'''
    #The dict contains a list of small df objects, so now concatenate them
    #by reassigning to the same dict, the memory footprint is not increasing 
    for chr_ in gen_matrix_df_list.keys():
        gen_matrix_df_list[chr_]=pd.concat(gen_matrix_df_list[chr_])

    '''Now, pipe each dataframe from the list using map.Pool() '''
    p = Pool(3)  # number of pool to run at once; default at 1
    result = p.map(matrix_to_vcf, list(gen_matrix_df_list.values()))
    p.close()
    p.join()