芹菜复杂的工作流程

Question

我正在尝试建立一个基于芹菜的工作流。我使用分组和和弦。

在下面的例子中，有独立的组([mytask1, mytask1, mytask1，…][-> myfinaltask1)，其中mytask1可以并行执行，但myfinaltask1应该在每个组之后调用。

代码:

def func1(date):
    subtasks = []
    for filepath in all_files:
        kwargs = {'date': date, 'hfile': filepath}
        subtask = mytask1.subtask(kwargs=kwargs)
        subtasks.append(subtask)
    chrd = chord(subtasks)
    chrdr = chrd(myfinaltask1.s(kwargs={'date': date}))
    return chrdr

def main(all_dates):
    subtasks = []
    for ad in all_dates:
        subtasks.append(func1(ad))
    g = group(subtasks)
    gr = g.apply_async()
    results = gr.get(propagate=False)  # sync wait!

main([2014, 2015, 2016])

异常:

File "/mypath/get_evi.py", line 265, in get_evi_year
    gr = g.apply_async()
File "/opt/venv/lib/python3.5/site-packages/celery/canvas.py", line 502, in apply_async
    type = self.type
File "/opt/venv/lib/python3.5/site-packages/celery/canvas.py", line 569, in type
    return self.app.tasks[self['task']]
File "/opt/venv/lib/python3.5/site-packages/celery/canvas.py", line 560, in app
    return self._app or (self.tasks[0].app if self.tasks else current_app)
AttributeError: 'bool' object has no attribute 'app'

我做错了什么?

Answer 1

您好像忘记把subtasks包装到group了。

def func1(date):
    subtasks = []
    for filepath in all_files:
        kwargs = {'date': date, 'hfile': filepath}
        subtask = mytask1.subtask(kwargs=kwargs)
        subtasks.append(subtask)
    chrd = chord(header=group(subtasks), body=myfinaltask1.subtask(kwargs={'date': date}))
    return chrdr