您是否将数据存储在与 Airflow 后端相同的数据库中？

如何解决您是否将数据存储在与 Airflow 后端相同的数据库中？

我正在尝试查找有关如何创建数据库、将数据存储在新创建的数据库中以及从该数据库中获取数据的信息，所有这些都来自 Airflow。我可以在网上找到的关于如何执行此操作的所有参考资料都只涉及 Airflow 的后端数据库。

我对 Airflow 有点陌生，我的理解是后端数据库是 Airflow 用来存储和访问与其自身操作相关的元数据的东西，而不是任务层数据。如果我想访问 sqlite 数据库以在整个 DAG 中存储数据，我可以找到一个 python 库来为我做这件事，但我觉得这不是通过 Airflow 访问数据库的传统方式。 (1) 创建 sqlLite 数据库，(2) 将数据存储在数据库中，以及 (3) 在气流中从该数据库中获取数据以供任务使用，而不是由任务使用的元数据的传统方法是什么？气流本身？

编辑：

我在评论中被要求提供有关我在 Airflow 上的工作流程的更多详细信息。此 DAG 的目标是通过数据处理管道从源移动数据，此过程的一部分涉及创建新的数据库和表（如果尚不存在），如下所示：

fetch data >> transform data >> put data elsewhere >> send alerts based on data

fetch data 和 put data elsewhere 步骤涉及将数据复制到本地数据库，我正在尝试找出在 Airflow 中执行此操作的“正确”方法。

解决方法

确实，您不得使用 Airflow 元数据数据库来存储您的数据。

首先需要在Airflow中为你要使用的Sqlite添加一个连接。要创建数据库，您可以使用 SqliteOperator： https://airflow.apache.org/docs/apache-airflow/1.10.13/_modules/airflow/operators/sqlite_operator.html

然后，您必须下载数据，这里有几个选项：

• 如果您使用的是低于 2.0 的 Airflow 版本，请使用 PythonOperator 或 BashOperator
• 对于 Airflow > 2.0，您可以使用 TaskFlow https://airflow.apache.org/docs/apache-airflow/stable/concepts.html#taskflow-api

无论哪种方式，您都必须编写如何下载数据的代码。您应该使用 sqlite_hook，方法是 get_pandas_df（或 get_records）

之后，无论是使用 taskflow 还是 pythonoperator，您都可以将数据上传到您想要的位置

总而言之，您总是希望搜索能够生成您想要的内容或编写 Python 脚本以与 PythonOperator、BashOperator 或 TaskFlow 一起运行的运算符。