3、第一个自定义Operator：继承BaseOperator、实现execute方法、最小化Operator示例。

好，咱们直接动手。

前面两章讲了 Airflow 的架构和 Operator 的定位，说白了就是告诉你：Operator 是 DAG 里的执行单元。那怎么自己写一个？这一章我们就来搞一个最小化的自定义 Operator。

我刚开始学的时候，也翻过不少源码。说实话，Airflow 的 BaseOperator 代码量不小，但你别被吓到。真正必须你实现的东西，其实就一个方法——execute。

3.1 最小化 Operator 长什么样？

先看代码，再解释。

from airflow.models import BaseOperator
from airflow.utils.decorators import apply_defaults

class HelloWorldOperator(BaseOperator):

    @apply_defaults
    def __init__(self, name: str = "World", **kwargs):
        super().__init__(**kwargs)
        self.name = name

    def execute(self, context):
        print(f"Hello, {self.name}!")
        return f"Greeted {self.name}"

嗯，就这么几行。我带你拆开看。

  • 继承 BaseOperator：这是必须的。所有自定义 Operator 的根都在这里。
  • @apply_defaults：这个装饰器很重要。它会把 DAG 级别的 default_args 自动合并到你的 Operator 参数里。我见过有人漏掉它，结果参数传不进去，排查了半天。
  • __init__：接收你的自定义参数。记得调用 super().__init__(**kwargs)，把剩下的参数交给父类处理。
  • execute(self, context)：这是核心。Airflow 在运行 Task 时，会调用这个方法。context 是一个字典，包含了当前 Task 实例、DAG Run、执行日期等信息。

核心原则：你只需要告诉 Airflow「做什么」，剩下的调度、重试、日志，BaseOperator 都帮你搞定了。

3.2 execute 方法到底能干什么？

你可能会问：execute 里面只能打印吗？当然不是。它可以做任何 Python 能做的事。

我在项目中遇到过这样一个场景：需要从某个内部 API 拉取数据，然后写入 S3。我当时就在 execute 里直接写了 requests 调用和 boto3 上传。后来发现这个逻辑在多个 DAG 里重复，才把它抽成了自定义 Operator。

execute 的返回值会被 Airflow 记录到 task_instance.xcom_pull 里，供下游 Task 使用。这是 Operator 之间传数据的主要方式。

小技巧：如果你想让 Operator 更灵活，可以把 API 地址、文件路径这些参数都放到 __init__ 里，而不是硬编码在 execute 中。

3.3 使用这个 Operator

写好了怎么用？跟内置 Operator 一模一样。

from airflow import DAG
from datetime import datetime
from your_package.operators import HelloWorldOperator

with DAG(
    dag_id="hello_world_dag",
    start_date=datetime(2024, 1, 1),
    schedule_interval=None,
    catchup=False,
) as dag:

    greet_task = HelloWorldOperator(
        task_id="say_hello",
        name="Airflow",
    )

你看，没有任何特殊之处。这就是继承 BaseOperator 的好处——你的 Operator 天然支持所有 Airflow 特性：重试、超时、池、SLAs 等等。

3.4 避坑指南

我曾经踩过一个坑：在 __init__ 里做了耗时操作，比如读取配置文件、建立数据库连接。结果 DAG 解析时变得巨慢，因为 Airflow 在解析 DAG 文件时会实例化所有 Operator。记住，__init__ 只负责接收参数，真正的逻辑放到 execute 里。

阶段 应该做的事 不应该做的事
__init__ 接收参数、校验参数类型 连接数据库、读取大文件、发起网络请求
execute 执行业务逻辑、处理数据、调用外部系统 修改 Operator 实例状态（不可重入）

注意：execute 方法必须是幂等的。如果 Task 失败重试，它会再次执行。如果你的操作不是幂等的（比如插入重复数据），记得在逻辑里做去重或使用 Airflow 的 retry 机制配合处理。

3.5 知识体系图

下面这张图帮你理清自定义 Operator 的核心结构：

自定义Operator核心结构图 自定义 Operator 核心结构 BaseOperator（基类） 你的自定义 Operator 继承 BaseOperator + 实现 execute __init__ 接收参数 调用 super().__init__ execute(context) 核心业务逻辑 返回 XCom 值 @apply_defaults 合并 default_args 参数自动注入 💡 最小化原则 __init__ 只做参数接收，execute 只做业务逻辑 保持幂等，避免在 __init__ 中做耗时操作

3.6 总结一下

这一章我们做了三件事：

  • 写了一个最小化的自定义 Operator
  • 理解了 execute 方法的职责
  • 知道了 __init__ 和 execute 各自该干什么

说白了，自定义 Operator 就是「把一段 Python 逻辑包装成一个 Airflow 认识的 Task」。你不需要关心调度、重试、日志——这些 BaseOperator 都替你扛了。

← 上一章 📖 返回目录 下一章 →