How to write in parallel to file with joblib? JoinableQueue problem

user_1397946 0
Viewed 1333

I am trying to write to a single file results of computations that are run over 100k+ files. Processing of a file takes ~1s and writes a single line to the output file. The problem itself is "embarrassingly parallel", I am only struggling with writing properly to a file (say, CSV). Here's what worked for me long time ago (Python 3.4?):

import os
from multiprocessing import Process, JoinableQueue
from joblib import Parallel, delayed

def save_to_file(q):
    with open('test.csv', 'w') as out:
        while True:
            val = q.get()
            if val is None: break
            out.write(val + '\n')
        q.task_done()

def process(x):
    q.put(str(os.getpid()) + '-' + str(x**2))

if __name__ == '__main__':
    q = JoinableQueue()
    p = Process(target=save_to_file, args=(q,))
    p.start()
    Parallel(n_jobs=-1)(delayed(process)(i) for i in range(100))
    q.put(None) 
    p.join()

Today (on Python 3.6+) it produces the following exception:

joblib.externals.loky.process_executor._RemoteTraceback: 
"""
(...)
RuntimeError: JoinableQueue objects should only be shared between processes through inheritance
"""

How to correctly write to a single file with joblib?

joblib python-3.x multiprocessing python
1 Answers

Turns out one way to achieve the task is through the multiprocessing.Manager, like this:

import os
from multiprocessing import Process, Manager
from joblib import Parallel, delayed

def save_to_file(q):
    with open('test.csv', 'w') as out:
        while True:
            val = q.get()
            if val is None: break
            out.write(val + '\n')

def process(x):
    q.put(str(os.getpid()) + '-' + str(x**2))

if __name__ == '__main__':
    m = Manager()
    q = m.Queue()
    p = Process(target=save_to_file, args=(q,))
    p.start()
    Parallel(n_jobs=-1)(delayed(process)(i) for i in range(100))
    q.put(None)
    p.join()

We let the Manager manage the context, rest remains the same (short of using normal Queue in place of JoinableQueue).

If anyone knows nicer / cleaner way, I will be happy to accept it as the answer.

user_13979460
Related