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

Question

我可以用 celery 吗Group原始作为 map /减少工作流程中的总括任务？

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

从文档:

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

这似乎意味着任务都发送给一名 worker ......

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

Answer 1

我发现可以使用 Chords 来解决类似 map 减少的问题。

@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)

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

总之:是的，这是可能的。感谢蔬菜大佬!