芹菜:如何在任务之间组织复杂的数据流

Question

有没有办法控制传递给芹菜链中下一个任务的参数？例如，我有两个任务，它们都存在于最终结果中，第二个任务使用第一个任务结果数据的一部分来完成它的工作。下面是一些示例说明代码：

import celery
import random
worker = celery.Celery()
@worker.task
def first():
    return(
        random.randint(1,9), # argument for second call
        # just a bunch of data I want to see in results
        # but not needed for second call
        [random.randint(1,9) for _ in xrange(100)]
    )
@worker.task
def second(number):
    return [random.randint(1,9) for _ in xrange(number)]

第二个调用取决于第一个，因此它必须是一个链。但我希望最终结果也包括第一个数据，而不是通过第二个任务绕过它。所以最简单的方法是：

argument, partial1 = first.s().apply_async().get()
partial2 = second.s(argument).apply_async().get()
results = (partial1, partial2)

但文件说这是错误的方式。怎么做对？

Answer 1

你可以

这样做

(partial1, partial2) = (first.s() | second.s(return_value_from_first))().get()

只需通过管道将第一个任务的输入传输到第二个任务即可。

import celery
import random
worker = celery.Celery()
@worker.task
def first():
    return(random.randint(1,9), [random.randint(1,9) for _ in xrange(100)])
@worker.task
def second(return_value_from_first):
    number, partial1 = return_value_from_first
    return [partial1, random.randint(1,9) for _ in xrange(number)]