Python multiprocessing job to Celery task but AttributeError
Question: I made a multiprocessed function like this,

    
    
    import multiprocessing
    import pandas as pd
    import numpy as np
    
    def _apply_df(args):
        df, func, kwargs = args
        return df.apply(func, **kwargs)
    
    def apply_by_multiprocessing(df, func, **kwargs):
        workers = kwargs.pop('workers')
        pool = multiprocessing.Pool(processes=workers)
        result = pool.map(_apply_df, [(d, func, kwargs)
                for d in np.array_split(df, workers)])
        pool.close()
        return pd.concat(list(result))
    
    def square(x):
        return x**x
    
    if __name__ == '__main__':
        df = pd.DataFrame({'a':range(10), 'b':range(10)})
        apply_by_multiprocessing(df, square, axis=1, workers=4)  
        ## run by 4 processors
    

Above "apply_by_multiprocessing" can execute Pandas Dataframe apply in
parallel. But when I make it to Celery task, It raised AssertionError:
'Worker' object has no attribute '_config'.

    
    
    from celery import shared_task
    
    @shared_task
    def my_multiple_job():
        df = pd.DataFrame({'a':range(10), 'b':range(10)})
        apply_by_multiprocessing(df, square, axis=1, workers=4)  
    

It's error trace is like this,

    
    
      File "/Users/yong27/work/goldstar/kinmatch/utils.py", line 14, in apply_by_multiprocessing
        pool = multiprocessing.Pool(processes=workers)
      File "/usr/local/Cellar/python3/3.4.0/Frameworks/Python.framework/Versions/3.4/lib/python3.4/multiprocessing/context.py", line 118, in Pool
        context=self.get_context())
      File "/usr/local/Cellar/python3/3.4.0/Frameworks/Python.framework/Versions/3.4/lib/python3.4/multiprocessing/pool.py", line 146, in __init__
        self._setup_queues()
      File "/usr/local/Cellar/python3/3.4.0/Frameworks/Python.framework/Versions/3.4/lib/python3.4/multiprocessing/pool.py", line 238, in _setup_queues
        self._inqueue = self._ctx.SimpleQueue()
      File "/usr/local/Cellar/python3/3.4.0/Frameworks/Python.framework/Versions/3.4/lib/python3.4/multiprocessing/context.py", line 111, in SimpleQueue
        return SimpleQueue(ctx=self.get_context())
      File "/usr/local/Cellar/python3/3.4.0/Frameworks/Python.framework/Versions/3.4/lib/python3.4/multiprocessing/queues.py", line 336, in __init__
        self._rlock = ctx.Lock()
      File "/usr/local/Cellar/python3/3.4.0/Frameworks/Python.framework/Versions/3.4/lib/python3.4/multiprocessing/context.py", line 66, in Lock
        return Lock(ctx=self.get_context())
      File "/usr/local/Cellar/python3/3.4.0/Frameworks/Python.framework/Versions/3.4/lib/python3.4/multiprocessing/synchronize.py", line 164, in __init__
        SemLock.__init__(self, SEMAPHORE, 1, 1, ctx=ctx)
      File "/usr/local/Cellar/python3/3.4.0/Frameworks/Python.framework/Versions/3.4/lib/python3.4/multiprocessing/synchronize.py", line 60, in __init__
        kind, value, maxvalue, self._make_name(),
      File "/usr/local/Cellar/python3/3.4.0/Frameworks/Python.framework/Versions/3.4/lib/python3.4/multiprocessing/synchronize.py", line 118, in _make_name
        return '%s-%s' % (process.current_process()._config['semprefix'],
    AttributeError: 'Worker' object has no attribute '_config'
    

It seems that because Celery worker is not a normal process. How can I solve
this problem? I'm using Python3.4, Django 1.6.2, celery 3.1.10, django-celery
3.1.9, pandas 0.12.0.


Answer: I don't know why multiprocessing doesn't work, but I recommend you to use
celery group task.

    
    
    from celery import task, group
    
    def feeds_fetch(feeds):
        g = group(fetch_one.s(feed) for feed in feeds)
        g.apply_async()
    
    
    @task()
    def fetch_one(feed):
        return feed.fetch()
    

