directed-acyclic-graphs - Airflow 任务引用多个先前的任务？

Question

有没有办法让我的任务需要完成多个仍然能够独立完成的上游任务？

下载_fcr --> process_fcr --> load_fcr

下载_调查 --> process_survey --> 负载调查

create_dashboard 应该需要 load_fcr 和 load_survey 才能成功完成。

我不想强制“调查”任务链中的任何内容要求完成“fcr”任务链中的任何内容。我希望它们并行处理并且即使失败也能完成。但是，仪表板任务需要在开始之前完成加载到数据库。

fcr *-->*-->*
             \
               ---> create_dashboard
                /
survey *-->*-->*

Answer 1

您可以通过 任务列表到 set_upstream 或 set_downstream。在您的情况下，如果您特别想使用 set_upstream，您可以将您的依赖项描述为:

create_dashboard.set_upstream([load_fcr, load_survey])

load_fcr.set_upstream(process_fcr)
process_fcr.set_upstream(download_fcr)

load_survey.set_upstream(process_survey)
process_survey.set_upstream(download_survey)

看看 airflow's source code :即使您只将一个任务对象传递给 set_upstream，它实际上在执行任何操作之前都会围绕它包装一个列表。