airflow - 将运算符融合在一起

Question

我还在部署中Airflow我已经觉得有必要合并 operator一起。最常见的用例是将运算符与相应的 sensor 耦合。 .例如，可能想要将 EmrStepOperator 链接在一起。和 EmrStepSensor .

我正在创建我的 DAG小号 programmatically ，其中最大的一个包含 150 多个(相同的)分支，每个分支对不同的数据位(表)执行相同的一系列操作。因此将构成单个逻辑步骤的任务组合在一起在我的 DAG 中会有很大帮助。
以下是我的项目中的 2 个有争议的例子，为我的论点提供动力。
1、从S3路径中删除数据，然后写入新数据
此步骤包含 2 个运算符

DeleteS3PathOperator : 扩展自 BaseOperator & 使用 S3Hook

HadoopDistcpOperator : 扩展自 SSHOperator

2.有条件执行MSCK REPAIR在 Hive表
此步骤包含 4 个运算符

BranchPythonOperator : 检查 Hive 表是否分区

MsckRepairOperator : 扩展自 HiveOperator并对(分区)表执行 MSCK REPAIR

Dummy(Branch)Operator : 构成到 MsckRepairOperator 的备用分支路径(对于非分区表)

Dummy(Join)Operator : 组成两个分支的连接步骤

单独使用操作符当然可以提供更小的模块和更细粒度的日志/调试，但在大型 DAG 中，减少困惑可能是可取的。根据我目前的理解，有两种方法可以将运营商链接在一起

Hook s
在钩子(Hook)中编写实际的处理逻辑，然后在单个运算符中使用尽可能多的钩子(Hook)(在我看来当然是更好的方法)

SubDagOperator一个 risky和 controversial做事的方式；另外 SubDagOperator makes me frown 的命名约定.

我的问题是

运算符应该完全组合还是最好有离散的步骤？

任何陷阱，上述方法的改进？

还有其他将运算符组合在一起的方法吗？

在 Airflow 的分类学中，Hooks 的主要动机与上述相同，还是它们也有其他用途？

更新 1
3.多重继承
虽然这是 Python功能而不是 Airflow具体来说，值得指出的是，多重继承在组合运算符的功能时可以派上用场。 QuboleCheckOperator ，例如，已经使用它编写了。但是在过去，我尝试过这个东西来融合 EmrCreateJobFlowOperator 和 EmrJobFlowSensor ，但当时我遇到了 issues与 @apply_defaults decorator并放弃了这个想法。

Answer 1

我结合了各种钩子(Hook)来根据我的需要创建一个单一的运算符。一个简单的例子是我在钩子(Hook)中加入 gcs 删除、复制、列表方法和 get_size 方法来创建一个名为 GcsDataValidationOperator 的操作符。 .经验法则是 幂等性 即，如果您多次运行它应该产生相同的结果。

Should operators be composed at all or is it better to have discrete steps?

唯一的缺陷是可维护性，有时当主分支中的钩子(Hook)发生更改时，如果有任何重大更改，您将需要手动更新所有运算符(operator)。

Any pitfalls, improvements in above approaches?

您可以使用 PythonOperator并使用带有 .execute 的内置 Hook 方法，但它仍然意味着 DAG 文件中有很多细节。因此，我仍然会采用新的运营商方法

Any other ways to combine operators together?

Hooks 只是与外部平台和数据库(如 Hive、GCS 等)的接口(interface)，并构成运算符(operator)的构建 block 。这允许创建新的运算符。此外，这意味着您可以自定义模板化字段，在新运算符(operator)内的每个细化步骤上添加松弛通知，并拥有自己的日志记录详细信息。

In taxonomy of Airflow, is the primary motive of Hooks same as above, or do they serve some other purposes too?

FWIW:我是 PMC 成员，也是 Airflow 项目的贡献者。