порождать несколько процессов для записи разных файлов Python

Question

Идея состоит в том, чтобы написать N файлы с использованием N процессов.

Данные для файла, который будет написано приходят из нескольких файлов, которые хранятся на словарь, который имеет список в качестве значения, и это выглядит следующим образом:

dic = {'file1':['data11.txt', 'data12.txt', ..., 'data1M.txt'], 
     'file2':['data21.txt', 'data22.txt', ..., 'data2M.txt'], 
     ... 
     'fileN':['dataN1.txt', 'dataN2.txt', ..., 'dataNM.txt']} 

так file1 является data11 + data12 + ... + data1M и т.д .. .

Так что мой код выглядит следующим образом:

jobs = [] 
for d in dic: 
    outfile = str(d)+"_merged.txt" 
    with open(outfile, 'w') as out: 
     p = multiprocessing.Process(target = merger.merger, args=(dic[d], name, out)) 
     jobs.append(p) 
     p.start() 
     out.close() 

и merger.py выглядит следующим образом:

def merger(files, name, outfile): 
    time.sleep(2) 
    sys.stdout.write("Merging %n...\n" % name) 

    # the reason for this step is that all the different files have a header 
    # but I only need the header from the first file. 
    with open(files[0], 'r') as infile: 
     for line in infile: 
      print "writing to outfile: ", name, line 
      outfile.write(line) 
    for f in files[1:]: 
     with open(f, 'r') as infile: 
      next(infile) # skip first line 
      for line in infile: 
       outfile.write(line) 
    sys.stdout.write("Done with: %s\n" % name) 

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

Помощь!

Answer 1

Поскольку рабочие процессы запускаются параллельно основному процессу их создания, файлы с именем out закрываются до того, как работники могут писать им. Это произойдет, даже если вы удалите out.close() из-за инструкции with. Скорее передайте каждому процессу имя файла и дайте процессу открыть и закрыть файл.

Answer 2

Проблема заключается в том, что вы не закрываете файл в дочернем файле, так что данные с внутренней буферизацией теряются. Вы можете перенести файл на дочерний элемент или обернуть все это в блок try/finally, чтобы убедиться, что файл закрыт. Потенциальным преимуществом открытия родителя является то, что вы можете обрабатывать ошибки файла там. Я не говорю об этом убедительно, просто вариант.

def merger(files, name, outfile): 
    try: 
     time.sleep(2) 
     sys.stdout.write("Merging %n...\n" % name) 

     # the reason for this step is that all the different files have a header 
     # but I only need the header from the first file. 
     with open(files[0], 'r') as infile: 
      for line in infile: 
       print "writing to outfile: ", name, line 
       outfile.write(line) 
     for f in files[1:]: 
      with open(f, 'r') as infile: 
       next(infile) # skip first line 
       for line in infile: 
        outfile.write(line) 
     sys.stdout.write("Done with: %s\n" % name) 
    finally: 
     outfile.close() 

UPDATE

Там была некоторая путаница родителей/decriptors файлов дочерних и что происходит с файлами в ребенка. Основная библиотека C не очищает данные на диске, если файл все еще открыт, когда программа завершает работу. Теория состоит в том, что правильно запущенная программа закрывает вещи перед выходом. Вот пример, когда ребенок теряет данные, потому что он не закрывает файл.

import multiprocessing as mp 
import os 
import time 

if os.path.exists('mytestfile.txt'): 
    os.remove('mytestfile.txt') 

def worker(f, do_close=False): 
    time.sleep(2) 
    print('writing') 
    f.write("this is data") 
    if do_close: 
     print("closing") 
     f.close() 


print('without close') 
f = open('mytestfile.txt', 'w') 
p = mp.Process(target=worker, args=(f, False)) 
p.start() 
f.close() 
p.join() 
print('file data:', open('mytestfile.txt').read()) 

print('with close') 
os.remove('mytestfile.txt') 
f = open('mytestfile.txt', 'w') 
p = mp.Process(target=worker, args=(f, True)) 
p.start() 
f.close() 
p.join() 
print('file data:', open('mytestfile.txt').read()) 

Я запустить его на Linux, и я получаю

without close 
writing 
file data: 
with close 
writing 
closing 
file data: this is data