用于 map/reduce 工作流程的 Celery Group 任务

Question

我可以使用 Celery Group 原语作为 map/reduce 工作流程中的总括任务吗？

或者更具体地说：组中的子任务可以在多个服务器上的多个工作人员上运行吗？

来自文档：

However, if you call apply_async on the group it will send a special 
grouping task, so that the action of calling the tasks happens in a worker 
instead of the current process

这似乎意味着任务都发送给一个工人......

在 3.0 之前（并且仍然是），可以触发一个任务集中的子任务，该任务集中将在多个服务器上运行。问题是确定是否所有任务都已完成执行。这通常是通过轮询所有不是很优雅的子任务来完成的。 我想知道是否可以使用 Group 原语来缓解这个问题。

Answer 1

我发现可以使用 Chords 来解决类似地图缩减的问题。

@celery.task(name='ic.mapper')
def mapper():
    #split your problem in embarrassingly parallel maps 
    maps = [map.s(), map.s(), map.s(), map.s(), map.s(), map.s(), map.s(), map.s()]
    #and put them in a chord that executes them in parallel and after they finish calls 'reduce'
    mapreduce = celery.chord(maps)(reduce.s())    
    return "{0} mapper ran on {1}".format(celery.current_task.request.id, celery.current_task.request.hostname)

@celery.task(name='ic.map')
def map():
    #do something useful here
    import time
    time.sleep(10.0)
    return "{0} map ran on {1}".format(celery.current_task.request.id, celery.current_task.request.hostname)

@celery.task(name='ic.reduce')
def reduce(results):
    #put the maps together and do something with the results
    return "{0} reduce ran on {1}".format(celery.current_task.request.id, celery.current_task.request.hostname)

当映射器在由三个工作器/服务器组成的集群上执行时，它首先执行映射器来拆分您的问题并创建再次提交给代理的新子任务。这些并行运行，因为队列被所有代理使用。还创建了一个和弦任务，轮询所有地图以查看它们是否已完成。完成后，将执行 reduce 任务，您可以将结果粘合在一起。

总而言之：是的，这是可能的。感谢蔬菜大佬！