将任务结果输入芹菜中的地图

Question

我觉得这是一个非常简单的用例，但不是一个我从文档中成功解决的用例。

以下是我的问题的人为版本：

我有一个任务，它接受一个参数并返回一个列表。我想对列表中的每个元素应用另一个任务。然后（视情况而定）我可能希望继续处理其他任务。

chain(
    my.tasks.divide.s(data),
    my.tasks.conquer.map(),
    ).apply_async().get()

或者

chain(
    my.tasks.divide.s(data),
    my.tasks.conquer.map(),
    my.tasks.unite.s(),
    my.tasks.rule.s(),
    ).apply_async().get()

地图不是那样的，但我觉得应该是这样的！我只想从一个参数链接到一个列表，然后再返回。。这当然不难做到吗？

这里和这里的“解决方案”似乎过于复杂，我不敢相信这是最佳实践。

关于和弦、贴图和链的文档对这种特殊情况没有帮助。

编辑：到目前为止我有什么。这里的问题是，我必须调用。apply_async（）. get（）两次以获得实际结果，因为中间步骤返回一个和弦而不是实际结果：

@app.task()
def divide(item):
    return [x for x in item]

@app.task()
def conquer(item):
    return item.upper()

@app.task()
def rule(items):
    return "".join(items)

@app.task()
def conquer_group(items):
    return group([conquer.s(x) for x in items])

@app.task()
def rule_group(items):
    return chord([conquer.s(x) for x in items], rule.s())

def main():
    print chain(
        divide.s("foobarbaz"),
        rule_group.s(),
        ).apply_async().get().apply_async().get()

Answer 1

这里的问题是map（）需要一个迭代来实例化。apply_async（）但不能实例化接受链中最后一个任务结果的map（）签名。我认为实现这一点的简单方法是在任务中实例化map（）签名。

@app.task()
def divide(item):
    return [x for x in item]

@app.task()
def conquer(item):
    return item.upper()

@app.task()
def process(items):
    return conquer.map(items)

def run():
    (divide.s("foobar") | process.s()).apply_async()