13

我可以使用 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 原语来缓解这个问题。

4

1 回答 1

27

我发现可以使用 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 任务，您可以将结果重新组合在一起。

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

于 2012-10-15T14:13:23.827 回答