Airflow 3 — Operators, Sensors, Hooks & Providers

Bốn khối xây dựng của một task

Trong bài về DAG & Task ta đã thấy một DAG là tập các task có phụ thuộc. Nhưng một task cụ thể làm gì thì lại do một trong bốn khối xây dựng quyết định:

• Operator — template mô tả một loại công việc. Khi bạn khởi tạo một operator trong DAG, bạn tạo ra một task. Operator là "class", task là "instance".
• Sensor — một loại operator đặc biệt chuyên chờ một điều kiện xảy ra (file tới, partition sẵn sàng, task ở DAG khác xong).
• Hook — lớp bọc (wrapper) kết nối tới hệ ngoài (database, cloud storage, API). Operator hầu như luôn gọi Hook bên trong để làm việc thật.
• Provider — gói cài thêm (apache-airflow-providers-*) đóng gói operator/sensor/hook cho một nền tảng cụ thể (AWS, Google, Snowflake, dbt...).

Hiểu quan hệ giữa bốn khối này là chìa khoá để viết pipeline sạch và biết khi nào nên dùng cái có sẵn, khi nào nên tự viết.

Operator — template cho công việc

Một operator đóng gói logic của một hành động lặp lại được, tham số hoá để tái sử dụng. Airflow chia operator thành vài nhóm quen thuộc.

BashOperator & PythonOperator

Hai operator "nền" nhất, gần như luôn có mặt trong core:

• BashOperator — chạy một lệnh hoặc script shell.
• PythonOperator — gọi một hàm Python. Trong Airflow hiện đại, cách viết được ưa chuộng là decorator @task (TaskFlow API) thay cho việc khởi tạo PythonOperator trực tiếp; @task sinh ra một PythonOperator phía dưới nhưng cú pháp gọn hơn và xử lý truyền dữ liệu qua XCom tự nhiên hơn.

# (minh hoạ) Airflow 3.x — import từ Task SDK
from airflow.sdk import dag, task
from airflow.providers.standard.operators.bash import BashOperator
import pendulum

@dag(schedule="@daily", start_date=pendulum.datetime(2025, 1, 1), catchup=False)
def etl_demo():

    prepare = BashOperator(
        task_id="prepare_dir",
        bash_command="mkdir -p /data/landing/{{ ds }}",
    )

    @task
    def transform(run_date: str) -> int:
        # logic Python thuần; giá trị return đi vào XCom
        rows = do_transform(run_date)
        return rows

    prepare >> transform("{{ ds }}")

etl_demo()

Lưu ý Airflow 3: các operator "chuẩn" như Bash/Python đã tách ra provider standard (apache-airflow-providers-standard), nên đường import là airflow.providers.standard.operators.bash. Decorator @dag, @task và nhiều tiện ích tác giả DAG dùng nay đến từ Task SDK (airflow.sdk) — đây là thay đổi lớn về mặt import so với Airflow 2.

SQL operators — hợp nhất về SQLExecuteQueryOperator

Trước đây mỗi database có operator riêng (PostgresOperator, MySqlOperator, SnowflakeOperator...). Xu hướng hiện tại là SQLExecuteQueryOperator (trong provider common.sql) thay thế phần lớn chúng: bạn chỉ cần trỏ tới một Connection (qua conn_id) và viết SQL, còn việc dùng đúng Hook cho từng loại DB do Airflow tự lo. Các operator DB cũ phần lớn đã deprecated và trỏ ngầm về operator hợp nhất này.

# (minh hoạ) chạy SQL trên bất kỳ DB nào có Connection tương ứng
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator

load_dim = SQLExecuteQueryOperator(
    task_id="load_dim_customer",
    conn_id="dwh",                       # Connection trỏ tới warehouse
    sql="sql/load_dim_customer.sql",     # có thể là file (được templating)
)

Transfer operators

Nhóm operator chuyên di chuyển dữ liệu giữa hai hệ: S3ToRedshiftOperator, GCSToBigQueryOperator, LocalFilesystemToS3Operator... Tên theo mẫu AToBOperator. Chúng gói hai Hook (nguồn và đích) và luồng copy bên trong, giúp bạn không phải tự viết code stream dữ liệu.

KubernetesPodOperator & operator theo provider

• KubernetesPodOperator — chạy một task như một Pod độc lập trên Kubernetes, cách ly hoàn toàn về image và tài nguyên. Rất mạnh khi mỗi task cần môi trường/dependency riêng.
• Operator theo provider — mỗi nền tảng lớn có bộ operator riêng: SnowflakeCheckOperator, BigQueryInsertJobOperator, SparkSubmitOperator, các operator cho dbt, EMR, Databricks, HTTP, v.v.

Sensor — chờ một điều kiện

Sensor là operator đặc biệt: nó poll (kiểm tra lặp lại) một điều kiện cho tới khi điều kiện đúng, rồi mới cho các task phía sau chạy. Ví dụ: chờ file landing/2026-07-01/*.csv xuất hiện, chờ một partition Hive sẵn sàng, hay chờ một task ở DAG khác hoàn tất (ExternalTaskSensor).

Sensor có ba chế độ hoạt động, khác nhau ở cách chiếm tài nguyên khi chờ — đây là chủ đề dễ gây tốn tiền hạ tầng nếu chọn sai.

Chế độ poke

mode="poke" (mặc định): sensor giữ nguyên một worker slot trong suốt thời gian chờ. Cứ mỗi poke_interval giây nó kiểm tra điều kiện, giữa các lần kiểm tra nó không nhả slot mà chỉ ngủ. Đơn giản, phản ứng nhanh, nhưng nếu chờ lâu (hàng giờ) và có nhiều sensor cùng chờ thì bạn "khoá" nhiều slot cho việc ngồi không.

Chế độ reschedule

mode="reschedule": giữa hai lần poke, task nhả worker slot và chuyển về trạng thái chờ được lên lịch lại (up_for_reschedule). Đến lần kiểm tra kế tiếp, scheduler mới cấp lại slot. Tiết kiệm slot hơn hẳn khi chờ lâu, đổi lại có độ trễ lập lịch mỗi lần và tạo thêm bản ghi trạng thái. Nguyên tắc thực dụng: poke cho chờ ngắn (giây tới vài phút), reschedule cho chờ dài (phút tới giờ).

Deferrable — triggerer + asyncio

Cách hiện đại và tiết kiệm nhất là deferrable operator/sensor. Thay vì lặp poll đồng bộ, task được viết để defer (nhường quyền): nó đăng ký một trigger — một coroutine asyncio nhỏ — rồi giải phóng hoàn toàn worker slot. Trigger này chạy trên một tiến trình mới của Airflow gọi là triggerer. Một tiến trình triggerer duy nhất có thể phục vụ hàng nghìn trigger đồng thời nhờ vòng lặp sự kiện async. Khi điều kiện thoả, trigger "bắn" (fire) và task được đánh thức, tiếp tục trên worker.

So sánh nhanh ba chế độ:

Nhiều sensor có sẵn phiên bản deferrable (ví dụ tham số deferrable=True), và nhiều operator thường (không chỉ sensor) cũng có chế độ deferrable để chờ job bên ngoài hoàn tất mà không giữ slot.

# (minh hoạ) deferrable sensor chờ file landing trên S3
from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor

wait_file = S3KeySensor(
    task_id="wait_landing_file",
    bucket_name="my-lake",
    bucket_key="landing/{{ ds }}/orders_*.csv",
    wildcard_match=True,
    deferrable=True,       # nhả worker, dùng triggerer thay vì poll đồng bộ
    poke_interval=60,
    timeout=60 * 60 * 6,   # bỏ cuộc sau 6 giờ
)

Hook — cầu nối tới hệ ngoài

Một Hook là lớp bọc giao tiếp với một hệ thống bên ngoài: nó biết cách mở kết nối, xác thực, và cung cấp các phương thức tiện dụng (get_records, insert_rows, run...). Điểm mấu chốt: Hook không chứa credential cứng trong code. Nó đọc thông tin kết nối từ một Connection.

Connection

Một Connection là bản ghi định danh bằng conn_id, chứa host, port, login, password, schema và một trường extra (JSON) cho tham số riêng của từng loại. Connection được lưu ở metadata DB (mã hoá bằng Fernet key) hoặc — an toàn hơn trong môi trường production — ở một secret backend như AWS Secrets Manager, HashiCorp Vault, GCP Secret Manager. Nhờ đó:

• Đổi mật khẩu/endpoint không cần sửa code DAG.
• Cùng một DAG chạy được ở nhiều môi trường (dev/staging/prod) chỉ bằng cách đổi giá trị của conn_id.

Quan hệ Operator ↔ Hook ↔ Connection

Luồng điển hình: Operator nhận conn_id làm tham số → bên trong execute() nó khởi tạo Hook với conn_id đó → Hook đọc Connection để mở kết nối tới hệ ngoài. Bạn thường chỉ thấy Operator ở tầng DAG; Hook làm việc thầm lặng bên dưới. Nhưng khi viết @task hoặc custom code, bạn có thể gọi Hook trực tiếp:

# (minh hoạ) dùng Hook trực tiếp trong một @task
from airflow.sdk import task
from airflow.providers.postgres.hooks.postgres import PostgresHook

@task
def count_orders() -> int:
    hook = PostgresHook(postgres_conn_id="dwh")   # đọc Connection "dwh"
    rows = hook.get_records("SELECT count(*) FROM orders")
    return rows[0][0]

Providers — hệ sinh thái tách rời

Airflow core giữ nhỏ gọn (scheduler, executor, Task SDK, một ít operator qua provider standard). Mọi tích hợp với thế giới bên ngoài nằm trong các provider package — pip package tên apache-airflow-providers-<tên>.

Đặc điểm quan trọng của mô hình provider:

• Version độc lập với core. Provider amazon có thể lên bản mới (thêm operator, sửa lỗi) mà không cần nâng cấp Airflow core, và ngược lại. Điều này cho phép tích hợp bám sát API của nhà cung cấp.
• Chỉ cài cái bạn cần. pip install apache-airflow-providers-snowflake mang theo operator + hook + sensor + kiểu Connection cho Snowflake.
• Mỗi provider có thể đăng ký operator, hook, sensor, transfer, loại Connection mới, và cả các thành phần UI.

Vài provider hay dùng: standard (Bash/Python, các sensor cơ bản), common.sql (SQLExecuteQueryOperator), amazon (S3, Redshift, EMR, Glue), google (BigQuery, GCS, Dataproc), snowflake, apache.spark (SparkSubmitOperator), cncf.kubernetes (KubernetesPodOperator), http, dbt.cloud.

Custom operator & custom hook

Khi tích hợp có sẵn không đáp ứng, bạn tự viết. Nhưng hãy cân nhắc thứ tự ưu tiên trước:

1. Dùng @task (TaskFlow) cho logic đặc thù, dùng-một-lần trong một DAG. Nhanh, đọc như Python thường.
2. Viết custom Hook khi bạn cần tái sử dụng logic kết nối tới một hệ ngoài chưa có provider (một API nội bộ, một DB đặc thù).
3. Viết custom Operator khi bạn có một mẫu công việc tham số hoá được lặp lại qua nhiều DAG — đóng gói nó thành class để nhất quán và có template hoá tham số.

Viết custom Operator

Kế thừa BaseOperator và cài đặt phương thức execute(self, context) — đây là nơi công việc thật diễn ra khi task chạy. Khai báo template_fields để Airflow render Jinja (như {{ ds }}) cho các thuộc tính đó trước khi execute chạy.

# (minh hoạ) custom operator rút gọn — nạp file landing vào warehouse
from airflow.sdk import BaseOperator      # Airflow 3: BaseOperator từ Task SDK
from airflow.providers.postgres.hooks.postgres import PostgresHook

class LoadCsvToWarehouseOperator(BaseOperator):
    # cho phép Jinja: đổi ngày chạy trong đường dẫn khi render
    template_fields = ("source_path",)

    def __init__(self, *, conn_id: str, source_path: str, table: str, **kwargs):
        super().__init__(**kwargs)
        self.conn_id = conn_id
        self.source_path = source_path
        self.table = table

    def execute(self, context):
        hook = PostgresHook(postgres_conn_id=self.conn_id)   # dùng Hook, không hard-code cred
        self.log.info("Nạp %s vào %s", self.source_path, self.table)
        hook.copy_expert(
            sql=f"COPY {self.table} FROM STDIN WITH CSV HEADER",
            filename=self.source_path,
        )
        return {"table": self.table, "path": self.source_path}

Vài quy tắc:

• Không làm việc nặng trong __init__ — chỉ nhận tham số. Mọi thứ tốn tài nguyên/kết nối nằm trong execute, vì __init__ chạy cả lúc parse DAG.
• Dùng self.log để log; giá trị return của execute tự động vào XCom.
• Để hỗ trợ deferrable, gọi self.defer(trigger=..., method_name=...) trong execute, và cài đặt method callback.

Viết custom Hook

Kế thừa BaseHook, đọc Connection qua self.get_connection(conn_id), và cung cấp phương thức tiện dụng. Đặt logic kết nối ở Hook để nhiều operator/@task dùng chung, tránh lặp lại code xác thực.

Use case thực tế

1) Chờ file landing rồi nạp. Một pipeline nhận dữ liệu đối tác qua S3 hằng ngày. DAG bắt đầu bằng một deferrable S3KeySensor chờ file landing/{{ ds }}/*.csv (nhả worker suốt thời gian chờ, có thể lên tới hàng giờ mà không tốn slot). Khi file tới, một custom LoadCsvToWarehouseOperator (hoặc S3ToRedshiftOperator) nạp vào bảng staging. Vì file đường dẫn có {{ ds }}, ta khai báo nó trong template_fields để mỗi run trỏ đúng thư mục ngày.

2) Gọi SQL trên warehouse. Sau khi staging xong, một chuỗi SQLExecuteQueryOperator (trỏ conn_id="dwh") chạy các file SQL: dựng dimension, cập nhật fact, kiểm tra chất lượng. Nhờ operator hợp nhất, đổi warehouse (Postgres → Snowflake) chỉ cần đổi Connection và provider, không sửa cấu trúc DAG. Toàn bộ credential nằm trong secret backend, DAG chỉ tham chiếu conn_id.

Ghi nhớ

• Operator = template công việc; task = instance của nó. Ưu tiên @task cho logic Python một lần, operator có sẵn cho việc chuẩn.
• SQLExecuteQueryOperator (provider common.sql) đã thay thế phần lớn operator DB riêng lẻ — dùng nó qua conn_id.
• Sensor: poke (giữ slot, chờ ngắn) < reschedule (nhả slot theo lịch, chờ dài) < deferrable (nhả hẳn slot, triggerer + asyncio, tốt nhất khi chờ nhiều/lâu). Deferrable cần chạy tiến trình triggerer.
• Hook giữ logic kết nối và đọc Connection; credential nằm ở metadata DB (Fernet) hoặc secret backend — không hard-code trong DAG.
• Providers tách khỏi core, version riêng; chỉ cài gói cho nền tảng bạn dùng.
• Custom operator: kế thừa BaseOperator, làm việc trong execute(), khai báo template_fields; nhẹ hoá __init__.
• Airflow 3: nhiều import chuyển sang Task SDK (airflow.sdk cho @dag, @task, BaseOperator), và Bash/Python thuộc provider standard.

Đọc tiếp

• DAG & Task — nền tảng cấu trúc pipeline.
• TaskFlow & XCom — viết task bằng @task và truyền dữ liệu giữa các task.
• Executors & scaling — nơi task thực sự chạy và cách mở rộng, bao gồm vai trò triggerer.