跨 DAG 的气流持久数据存储

Question

我有几个 DAG 创建临时 AWS EMR 集群，然后在它们完成运行后终止它们。我想创建一个每天运行的新 DAG，并生成当天创建的每个 EMR 集群的报告以及运行时长，并通过电子邮件将此报告发送给不同的人。

我需要存储 EMR 集群 ID 值，以便我的报告生成器拥有当天每个 EMR 集群 ID 的列表。我想知道是否可以修改 Airflow 变量来存储此信息，例如，我可以有一个 Airflow 变量，其中键是“EMR_CLUSTERS”，值是 JSON 字符串，其中包含我想要记录的所有数据。或者我可以使用已经用于写入新表的 Airflow 元数据库，在那里我可以写入这些信息？

在 Airflow 中存储永久数据有哪些选择？

Answer 1

您提到的任何一个选项都可以：

1. 气流变量
2. 元数据数据库

第三个选项是网络存储。 如果您正在运行分布式气流，那么您可能将 DAG 存储在网络存储中，并将其安装到工作程序/调度程序/网络服务器中。在这种情况下，将基于文件的报告放在此存储上（并可能通过电子邮件发送等）将是一个可靠的选择。

You could write a plugin 可以与这 3 个中的任何一个一起使用，并且可以显示何时写入/发送的内容。

变量

Easily read/written，但是每天在 IMO 上覆盖它有点草率。

元数据数据库

使用SQLAlchemy 创建和读取/写入存储此信息的表。 您可以通过以下方式获得气流元数据数据库的会话：

from airflow import settings
session = settings.Session()

网络存储

在这种情况下，只需正常读取/写入文件即可。