Airflow - 任务中的并行执行答案

【问题标题】：Airflow - Parallel execution within a TaskAirflow - 任务中的并行执行
【发布时间】：2020-11-07 04:54:44
【问题描述】：

我有一个如下所示的数据库表：

name        last_successful
dataset1    2020-11-07 10:30
dataset2    2020-11-07 10:30

现在我想在气流中安排一个工作流，它执行以下操作：

从数据库中读取所有数据集name 和last_successful。
对于每个数据集，检查与下一个 30 分钟存储桶 last_successful + 30 min 对应的数据是否完成（假设这是一个黑盒子）
在 s3 数据分区中写入一个 _SUCCESS 文件，对应于在步骤 2 中找到的任何已完成的存储桶，并更新数据库中的 last_successful。

由于数据集彼此独立，因此可以对每个数据集并行化步骤 2+3。我怎样才能在气流中做到这一点？我不想为每个数据集创建单独的任务，因为数据集列表会不断增加。

【问题讨论】：

您说您不想创建单独的任务 - 是因为您不想手动添加新任务吗？可以动态创建任务，即基于 YAML 配置文件。
是的，不想手动操作。
是名称不变的数据集列表还是会发生许多变化（添加/删除等）？还有什么是启动新运行时基的触发器？事件库？
如果它受到许多变化的影响，这可能是个问题。 Airflow 可以处理动态的 dag/任务，但不是每次运行都会改变的那种。考虑到 Airflow 不断扫描 dag 文件，这意味着如果它需要从数据库中读取来执行此操作，您最终将使用打开的连接轰炸数据库。如果您愿意从文件（csv/json）中读取数据集名称，然后在数据库中查找它们，则可以解决此问题。
不，出于同样的原因。在运算符范围之外读取变量是不好的做法。您将轰炸气流的后端元存储。在下一个气流版本中，它还将为此类操作打印警告：github.com/apache/airflow/issues/11925 您必须避免从操作员范围之外的 DB（任何 DB）读取。因此我建议从文件中读取。添加新条目时，您只需将条目添加到文件中，如果您的 DAG 是动态的，它将自动为其创建一个流。

标签： python celery airflow

【解决方案1】：

总结关于cmets的讨论。

创建一个 DAG 作为解析从 DB 读取记录的一部分是一种不好的做法。 Airflow 不断解析文件，并打开与数据库的连接以获取最新记录。这意味着数据库上的连接负载很重。因此，Airflow 将为尝试使用此方法的用户添加警告（请参阅issue）。虽然警告出现在 Airflow Metastore 后端，但同样适用于任何其他数据库。

您可以通过在文件中列出数据集名称来处理您的用例。如果要创建具有动态任务的 DAG，则可以为文件中的任何新条目打开一个“分支”。然后，您可以在专用运算符中将该条目与您的数据库进行比较，以确保它是有效的。

方法可能类似于：

def get_file():
    with open('your_file') as f:
         lines = f.read().splitlines()
    return lines

file_list = get_file()
 with DAG(dag_id='my_dag',...
) as dag:
   start_op = DummyOperator(task_id='start_task')
   for dataset in file_list:
          my_op = MyOperator(task_id=dataset)
          start_op >> my_op

这样每次您将新数据集添加到文件时，它都会自动为其添加分支MyOperator

【讨论】：