python - 如何使用 Airflow 来编排简单的 pandas etl python 脚本？

Question

我喜欢 Airflow 的概念，但我停留在基础知识上。从昨天开始，我在 vm ubuntu-postgres 解决方案上运行了 Airflow 。我可以看到仪表板和示例数据 :)) 我现在想要的是迁移一个示例脚本，我用它来处理原始数据到准备好的数据。

假设你有一个 csv 文件的文件夹。今天我的脚本遍历它，将每个文件传递给一个列表，该列表将被转换为一个 df。之后，我准备他们的列名并进行一些数据清理并将其写入不同的格式。

1: pd.read_csv 用于目录中的文件

2:创建一个df

3:清理列名

4:干净的值(与 stp 3 平行)

5:将结果写入数据库

我将如何根据 Airflow 组织我的文件？脚本应该是什么样子？我是传递一个方法、一个文件还是必须为每个部分创建多个文件？在这一点上我缺乏基本概念:(我读到的关于 Airflow 的所有内容都比我的简单案例复杂得多。我正在考虑远离 Airflow 以及 Bonobo、Mara、Luigi，但我认为 Airflow 是值得的?!

Answer 1

我会使用 PythonOperator，将整个代码放入 Python 函数中，创建一个 Airflow 任务，仅此而已。

如果需要拆分这些步骤，也可以将 csv 文件的加载和数据库写入放在一个函数中。所有这些都将放在一个 DAG 中。

因此，您的一个 DAG 将执行以下三项任务:

loadCSV (PythonOperator)
parseDF (PythonOperator)
pushToDB (PythonOperator)

如果您使用多个任务，您需要使用Airflow's XCom .一开始只使用一项任务会更容易。

这里有几个代码示例在标签 Airflow 下。当你创造了一些东西，再问一遍。