distribute sequential python code using celery

celery

Solution

In your case you have to call `get_filters()`, wait for result and pass it to `get_customers`, `get_orders`, `get_returns` signatures inside `chord` with new function `merged` as callback.

Take a look at this:

def extract_customer():
    filt_a, filt_b = get_filters()

    result = chord([
        get_customers.s(filt_a, filt_b),
        get_orders.s(filt_a, filt_b),
        get_returns.s(filt_a, filt_b)
    ])(merged.s())

    result.get()

@shared_task
def merged(args):
    cust, ord, ret = args
    return cust.join([ord, ret])

The only thing which could be an issue here is what type of results you are getting from get_* functions and can they be send through celery broker. Either they should be simple objects which can be encoded with json, or use pickle for passing them between tasks.

Problem

I have the following tasks: ``` get_filters() # returns a list of filters get_customers(filter_a, filter_b) # returns a pandas DataFrame containing customers get_orders(filter_a, filter_b) # returns a pandas DataFrame of customers and aggregate purchase statistics get_returns(filter_a, filter_b) # returns a pandas DataFrame of customers and aggregate return statistics ``` The sequential code works like this: ``` def extract_customer(): filt_a, filt_b = get_filters() cust = get_customers(filt_a, filt_b) ord = get_orders(filt_a, filt_b) ret = get_returns(filt_a, filt_b) merged = cust.join([ord, ret]) ``` I would like to distribute the task with celery so that get_filters gets executed first, then get_customers, get_orders & get_returns execute concurrently. and finally, when they have finished executing, the merge function returns a merged dataset. I'm not sure how to do that using the canvas in celery. Thank you for your help.

Original source