代码之家 › 专栏 › 技术社区 › y2k-shubham

将操作员融合在一起

airflow

y2k-shubham · 技术社区 · 6 年前

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

我在创造我自己 DAG S programmatically ,其中最大的一个包含150+(相同) 分支机构 ,每个对不同的数据位(表)执行相同的一系列操作。因此 将构成单个逻辑步骤的任务组合在一起 在我的眼皮底下会有很大的帮助。

以下是我的项目中两个有争议的例子,为我的论点提供动力。

1。从S3路径删除数据,然后写入新数据

此步骤包括2个操作员

DeleteS3PathOperator :从 BaseOperator 及用途 S3Hook
HadoopDistcpOperator :从 SSHOperator

2。有条件地执行 MSCK REPAIR 在 Hive 桌子

此步骤包含4个运算符

BranchPythonOperator :检查配置单元表是否分区
MsckRepairOperator :从 HiveOperator 并在上执行msck修复( 分型的 表
Dummy(Branch)Operator 化妆 备用分支路径 到 摩根士丹利资本公司 (对于非分区表)
Dummy(Join)Operator :弥补 连接步骤 对于两个分支

在中使用运算符隔离当然可以提供更小的模块和更多的 细粒度的 日志记录/调试,但在大DAG中,减少混乱可能是可取的。根据我目前的理解,有两种方法可以将操作员连接在一起

Hook S

写实 处理逻辑 在钩子中,然后在一个操作符中使用任意多的钩子(我认为这当然是更好的方法)
SubDagOperator

一 risky 和 controversial 做事情的方式;另外,子数据运算符的命名约定 makes me frown .

我的问题是

运算符应该是组成还是最好有 离散的 步骤?
上述方法是否存在缺陷和改进?
有其他方法吗结合操作员在一起?
在 分类法 对于气流,钩子的主要动机是和上面一样,还是它们也有其他用途?

1 回复 | 直到 6 年前

kaxil 6 年前

我结合了各种钩子,根据我的需要创建了一个单一的操作员。一个简单的例子是,我在hook中使用clubbed gcs delete、copy、list方法和get-size方法来创建一个名为 GcsDataValidationOperator . 经验法则是幂等也就是说,如果您运行多次,它将产生相同的结果。

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

唯一的缺陷是可维护性,有时当主分支中的钩子发生变化时,如果有任何中断的变化,您需要手动更新所有的操作员。

上述方法是否存在缺陷和改进?

你可以使用 PythonOperator 使用内置的钩子 .execute 方法,但它仍然意味着DAG文件中有很多细节。因此,我仍然会选择一种新的操作员方法

有没有其他方法可以将运算符组合在一起?

钩子只是到外部平台和数据库(如hive、gcs等)的接口,并为操作员形成构建块。这允许创建新的运算符。此外,这意味着您可以自定义模板化字段,在新操作员的每个细化步骤上添加松弛通知,并拥有自己的日志记录详细信息。

在气流分类中,钩子的主要动机与上面的相同,还是它们也有其他用途?

我是项目管理委员会成员,也是气流项目的贡献者。