2.2 Operator与任务集成

2.2 Operator 与任务集成

本节摘要：Operator 不是魔法函数名，而是运行时契约的翻译器：它向调度器承诺可重试、可记日志，向执行器声明资源与进程形态，向目标系统通过 Connection 索取坐标。Hooks 管连接，Operator 管步骤，Connection 管凭证。三角任一角缺失，界面上的绿图也会在第一次真正访问外部系统时断裂。断裂按层报，不要混成一句任务报错。

核心问题

阅读完本节，你应当能够：

1. 画出 Connection、Hook、Operator 的职责边界
2. 判断何时用专用 Operator，何时收口到 PythonOperator
3. 把主机和密码从代码里挪到 conn_id
4. 解释“本机能 import，Worker 上 ModuleNotFound”为何是集成问题而不是调度问题

一、一次任务调用签了三份约

原文把 PostgresOperator 这类写法拆成三重契约。对调度器：任务失败可重试、超时可观察、不会假装跨任务事务。对执行器：可能要特定队列、池、容器镜像。对数据库：postgres_conn_id 指向已登记的连接，包含认证与可达性。你在笔记本上跑通 SQL，只验证了第三份约的一半——还没验证 Worker 网络策略和库版本。

extract = PostgresOperator(
    task_id="extract_users",
    sql="SELECT id FROM users WHERE updated_at >= '{{ ds }}'",
    postgres_conn_id="postgres_analytics",
)
​

模板 ds 来自运行窗口，不是执行那一刻的日历。专用 Operator 的价值是：SQL 文本、连接 id、日志钩子都按社区约定放好，审计时能搜到“哪一个 conn_id”。等价的 PythonOperator 里手写客户端，容易把密码取出来印到日志，或忘记关闭连接。

通用 PythonOperator 适合没有现成 Provider、或逻辑是多步胶水的情况。代价是契约要自己补：超时、重试哪些异常、日志里打不打敏感字段。BashOperator 能调脚本，但把业务逻辑藏进 shell 会让测试和权限审查变难；能用 Python 表达的，我更倾向不进 bash。

二、三角闭环：Connection、Hook、Operator

Connection 存在元数据（或密钥后端）里，有 conn_id、类型、主机、登录、加密密码、extra。Hook 按类型读取 Connection，提供 get_conn 这类方法。Operator 在 execute 里调 Hook。这是集成基础设施栈。新增一个云存储任务，正确顺序是：登记连接、确认 Worker 能解析该主机、在任务里只写 conn_id。

Connection 存坐标
     │
     ▼
  Hook 建立会话
     │
     ▼
 Operator 执行这一步业务
​

凭证轮转必须假设 DAG 代码不改。密码只活在连接对象里，Worker 运行时拉取。原文举过离职员工删了 UI 账号、连接仍留在表里的例子：认证边界被 Connection 读权限绕开。所以能看连接列表的角色，不一定能看明文密码；2.x 之后把读连接与读密码拆得更细，就是为了这件事。

依赖库安装在执行侧。Scheduler 解析 DAG 时可能只 import 类名；真正 import snowflake.connector 发生在 Worker。本机开发环境装了包，生产镜像没装，就会出现“解析成功、运行 ModuleNotFound”。Kubernetes 执行器下，包在任务镜像里，不在调度器镜像里。Celery 则要求该队列上的 Worker 一致。集成清单应写成：Provider 版本、系统库、网络出口、连接 id，四件套随 DAG 一起评审。

Sensor 也是 Operator 家族：周期性 poke 外部条件。传统实现占着 Worker 槽位睡觉。等待 EMR 集群、等待对象存储文件，用可延迟版本能把槽位还回去，原文提到资源利用率可显著上升。选型直觉：等待超过一分钟、并发等待很多，优先可延迟。

自定义 Operator 只在三角无法表达时再写。要遵守 execute 签名、把连接走 Hook、把大结果写外部存储。不要在自定义类里硬编码生产主机。插件加载路径放到第 5 章；本章只要求：自定义也是契约翻译器，不是复制粘贴函数的新地方。

图 集成故障常断在哪一层

三、工程上怎么选、怎么验

我选算子的顺序：有官方 Provider 且参数够用，用专用 Operator；逻辑超过三步分支，用 Python 函数并保持 Hook 取连接；必须调遗留二进制，才用 Bash，并把可执行文件锁进镜像。跨云的两套凭证分成两个 Connection，不要在 extra 里塞两套密码。

验证不要只点 Trigger。用一个只读查询任务验证连接；用故意错误的 conn_id 确认失败信息是否泄露密码；在与生产相同的镜像里跑一次，而不是在笔记本虚拟环境。模板语法在专用 Operator 的 sql/bash_command 里生效，在普通 Python 函数参数里不会自动渲染——这是另一类“本机能 print、线上是花括号原文”的坑。

⚠️ 常见坑：在 Python 函数里打印 Connection 对象或 extra JSON。日志会被收集到集中平台，等于把密钥复制到第二套存储。
💡 关键直觉：界面 DAG 变绿只说明解析通过。集成是否成立，要看 Worker 那一侧的包、网、证。

Provider 版本要跟核心版本一起记。社区维护了大批云与数据工具插件，这是 Airflow 相对自研调度的真实护城河。升级核心却锁死旧 Provider，会出现类路径搬家。集成测试应 import 真实 Operator 类，而不是只检查 DAG 字典结构。

空任务（EmptyOperator 或旧 Dummy）适合做汇聚点：多条分支成功后经过一个锚再往下。它不传数据。需要把多路上游的路径合并，仍要各自写存储，再在汇聚后的 Python 任务里读清单。

四、接一个新系统的验收顺序

不要从写 Operator 参数开始。第一，登记 Connection，类型与主机用非生产先打通。第二，在与 Worker 相同的镜像里确认 SDK 能 import。第三，写一个只读任务，超时设短，确认错误信息没有把密码打出来。第四，再写真正的抽取或装载，SQL 或对象路径用窗口模板，不用 now。第五，把 conn_id、Provider 名、镜像变更单写进 DAG 的 doc_md。少任一步，后面的“偶发失败”都会变成三方扯皮。

原文场景值得当验收反例：缺连接、缺 Snowflake 包、过期私钥，三次失败不在同一层。验收清单按层打勾，出问题先报层，再报异常类名。网络策略变更（Worker 不能出网）属于系统契约，不是调度器 bug。TLS 证书更新同理。集成栈要能容忍证书轮转：密码在保险库，连接对象在运行时取，DAG 代码零修改。

BashOperator 的验收还要多两问：二进制在不在镜像 PATH，脚本是否假设交互式 tty。PythonOperator 多一问：函数里有没有悄悄 new 客户端而不走 Hook。专用 Operator 多一问：模板字段会不会渲染，普通参数会不会原样留下花括号。

空任务做汇聚时，验收看的是规则不是 SQL。多路并行后经过 EmptyOperator 再往下，EmptyOperator 默认仍要上游成功。若其中一路是可选告警，规则要改，否则空任务永远等那个 skipped 的告警。这是集成问题也是依赖问题，放在本节是因为很多人会以为空任务“什么都不做所以怎么都能过”。

问题：Hook 能在解析期调用吗？

不要。解析期拿连接会把元数据库和可能的密钥后端打进扫描循环，也容易在调度器所在网络去碰业务库。Hook 留在 execute。需要在解析期决定图结构，用静态配置，不要用 Connection 里的主机列表去动态长节点。

问题：一个 conn_id 给开发和生产共用，靠 extra 切换，可以吗？

不要。conn_id 应按环境分开，发布时环境注入不同值，或不同 Airflow 实例登记同名但指向不同主机。靠 extra 切换等于把环境判断写进每一份代码，漏一次就是生产事故。集成的清晰来自抽屉名字稳定、内容按环境变，而不是名字玩花样。

五、Provider 升级时的集成回归

核心小升、Provider 大升，导入路径可能搬家。回归最小集：CI 里 import 所有生产用到的 Operator 类；预发跑探针乙只读连接；抽一条真实 DAG 的非写任务走一遍。不要只看 changelog 里的“兼容”。兼容往往指类还在，不指默认参数语义没变。连接 extra 字段更是重灾区。升级窗口与 DAG 发布窗口错开，避免同一天既换包又改 SQL，出问题无法二分。集成栈的版本矩阵应写在平台清单，作者不在 DAG 文件里私自假设。缺包的症状记得分流：解析期失败是调度器侧缺 import，运行期 ModuleNotFound 是 Worker 侧缺包，两者修的镜像不同。

六、日志里不许出现的集成痕迹

连接串、密码、token、完整带个人信息的 WHERE 条件、私钥片段、保险库响应体，出现任一项即事故。Operator 的日志钩子默认可能打出 SQL，要在平台层或封装层打码。Python 里 print 异常对象也可能带请求头。集成规范因此包含日志规范，不只包含 conn_id 规范。验收探针乙应故意用错误密码，确认报错文本不含密码本身。含了，先修日志再接通。Worker 与调度器的日志收集若未脱敏，集中日志平台会变成第二凭证库，权限模型瞬间失效。5.2 会谈 RBAC，但 2.2 不把秘密打进日志，RBAC 才有意义。意义建立在秘密还在抽屉里。抽屉的钥匙是 conn_id。钥匙可以出现在代码里，锁芯内容不行。这是集成与安全的接缝，放在本节是因为漏点几乎总在第一次写 Operator 的时候，而不是在后来补防火墙的时候。

七、连接抽屉的生命周期

创建：环境名在 conn_id 里可区分，或实例隔离。轮转：只改保险库或连接密码字段，不改 DAG。移交：离职时改 owner 与权限，不删抽屉除非确认无任务引用。删除：先搜引用，再停任务，再删。四段缺一，就会出现“密码换了但旧任务还在用缓存”或“抽屉删了界面全红却不知道谁引用”。生命周期写进平台操作，不写进每个 DAG。DAG 只写 conn_id。id 稳定，内容可变。稳定是集成的第一原则。可变是安全的第一原则。两原则靠生命周期调和。调和失败的典型是开发生产共用 id。共用让轮转变成俄罗斯轮盘：改一处，两套环境一起变。一起变则无法灰度。无法灰度则不敢轮转。不敢轮转则密码永不过期。永不过期的密码是 5.2 的噩梦，根子在 2.2 的抽屉命名。命名在创建那一天就定了。定错，后面所有安全章都在还债。还债很贵。创建时多写一个环境后缀，便宜。便宜的事放在本节，是因为第一次登记连接就发生在这里。这里不慎，后面全是补丁。补丁补不了共用 id 的原罪。原罪只能迁移 id。迁移 id 等于改所有 DAG。所有 DAG 一起改，无法二分。无法二分的变更，4.1 已经反对过。反对要从创建开始。开始在 2.2。

本章回顾

• Operator 翻译三份契约：调度可观察、执行可落地、外部系统可认证
• Connection 存坐标，Hook 开会话，Operator 做事，密码不进 DAG
• 包安装在 Worker 镜像，解析成功不等于运行能 import
• 专用算子优先，Python 其次，Bash 垫底，自定义算子最后
• Sensor 会占槽位，长等待用可延迟实现
• 模板渲染只发生在声明会渲染的字段，普通函数参数不会自动换 ds

下一节处理工位之间的边、纸条和触发规则。