airflow - Apache Airflow 最佳实践 : (Python)Operators or BashOperators

Question

这些天我正在从事一个新的 ETL 项目，我想尝试将 Airflow 作为作业经理。
我和我的同事都是第一次在 Airflow 上工作，我们采用了两种不同的方法:我决定编写 python 函数(类似于 apache-airflow 项目中包含的操作符)，而我的同事使用 Airflow 来调用外部 python 脚本通过 BashOperator。

我想知道是否有类似“良好做法”的东西，如果这两种方法同样好，或者我应该考虑一种而不是另一种。

对我来说，主要区别是:
- 使用 BashOperator，您可以使用具有特定包的特定 python 环境调用 python 脚本
- 使用 BashOperator，任务更加独立，如果 Airflow 变得疯狂，可以手动启动
- 使用 BashOperator 任务间通信有点难以管理
- 使用 BashOperator 任务错误和失败更难管理(bash 任务如何知道任务失败或成功之前的任务？)。

你怎么认为？

Answer 1

在这些情况下，我个人更喜欢使用 PythonOperator 而不是 BashOperator。这是我所做的以及为什么:

包含我所有 DAG 的单个存储库。这个 repo 还有一个 setup.py这包括作为依赖的 Airflow ，以及我的 DAG 需要的任何其他东西。 Airflow 服务从安装这些依赖项的 virtualenv 运行。这将处理您提到的有关 BashOperator 的 python 环境。

我尝试将所有与 Airflow 无关的 Python 逻辑放在它自己的外部打包的 Python 库中。该代码应该有自己的单元测试，也有自己的主要内容，因此可以在独立于 Airflow 的命令行上调用它。这解决了您关于 Airflow 何时发疯的观点!

如果逻辑足够小以至于无法分离到自己的库中，我会将它放在我的 DAG 存储库中的 utils 文件夹中，当然还有单元测试。

然后我在 Airflow 中使用 PythonOperator 调用此逻辑。与 BashOperator 模板脚本不同，python 可调用文件可以轻松进行单元测试。这也意味着您可以访问诸如启动 Airflow DB session 、将多个值推送到 XCom 等内容。

就像您提到的那样，使用 Python 处理错误会更容易一些。您可以轻松捕获异常并检查返回值。您可以选择使用 raise AirflowSkipException 将任务标记为已跳过。 .

仅供引用 BashOperator，如果脚本退出并显示错误代码，Airflow 会将任务标记为失败。