airflow - 我如何编写一个具有与执行日期相关的动态任务集的 DAG？

Question

我们有一个 DAG，它从广告平台中提取一些数据。这些广告被组织成事件。我们的目标是为这些事件引入高级指标。为此，我们首先需要获取给定执行日期的事件事件列表——幸运的是，只要我们知道我们想要查询的时间范围，广告平台的 API 就可以让这变得微不足道。

目前，我们的 DAG 的结构是去获取这些事件，然后将它们存储在 S3 中，最后存储在 Redshift 中。然后，我们在设置为每个事件提取数据的后续任务之前查询 Redshift。这是最重要的部分。我们也可以在 S3 中查看，但问题是键是用 ds 宏的值模板化的。在构建 DAG 本身时，似乎没有办法知道该值。

我们当前的方法也不知道执行日期，因此它总是查询所有事件，即使这些事件在我们感兴趣的时间段内不活跃也是如此。

为了更具体一点，DAG 现在看起来像这样:

另一种方法是将所有这些汇总到一个运算符中，该运算符封装获取当前执行日期的一组事件，然后获取每个事件的指标。我们避免了这种情况，因为这似乎排除了通过每个事件的单独任务并行提取数据。

我们如何编写此 DAG，以维持通过动态查询事件的 Redshift 表提供的并行化，但事件被正确地限制在执行日期？

Answer 1

我不认为这是可能的。 DAG 只能在 DAG 的 python 定义定义的一种配置中呈现。您将无法根据执行日期来控制 DAG 的哪个版本呈现，因此您将无法回顾 DAG 在过去应该如何呈现，例如。如果您希望当前 DAG 根据执行日期呈现，那么您可以在 DAG 的 python 定义中编写一些逻辑。

根据您编排 Airflow 作业的方式，您可以像您描述的那样拥有一个运算符(operator)，但让该运算符(operator)在 Redshift 上启动并行查询并在所有查询完成时终止。