Bài 3

Operators, Sensors & XCom

Đi sâu vào "ngữ pháp" của Airflow: các Operator để định nghĩa việc cần làm, Sensor để chờ điều kiện, XCom để truyền dữ liệu nhỏ giữa task, cùng Hooks, Connections, Variables để quản lý kết nối và cấu hình, và branching để rẽ nhánh động.

Operators phổ biến

Mỗi operator = một loại việc

Operator là khuôn mẫu cho task. Một số có sẵn trong core, số khác đến từ provider package (cài thêm, ví dụ apache-airflow-providers-postgres).

Operator	Việc làm	Nguồn
BashOperator	Chạy lệnh shell / script bash.	Core
PythonOperator	Gọi một hàm Python (`python_callable`).	Core
EmailOperator	Gửi email (thông báo, cảnh báo).	Core
PostgresOperator	Chạy câu lệnh SQL trên Postgres (dùng một Connection).	Provider
SparkSubmitOperator	Submit một Spark job lên cụm Spark.	Provider
SimpleHttpOperator	Gọi một HTTP/REST API.	Provider

Quy tắc vàng: chọn operator chuyên dụng thay vì nhét mọi thứ vào PythonOperator. Việc nặng (truy vấn lớn, xử lý batch) nên giao cho PostgresOperator/SparkSubmitOperator — Airflow chỉ điều phối.

Sensors — chờ điều kiện

poke vs reschedule

Sensor là loại operator đặc biệt: nó chờ cho tới khi một điều kiện thành true rồi mới cho task sau chạy. Ví dụ chờ file xuất hiện, chờ task ở DAG khác xong, chờ object trên S3.

Sensor	Chờ gì
FileSensor	Chờ một file/thư mục xuất hiện trên hệ thống file.
ExternalTaskSensor	Chờ một task cụ thể của một DAG khác hoàn thành.
S3KeySensor	Chờ một object (key) xuất hiện trong bucket S3.

Sensor có hai mode:

Mode	Cơ chế	Đánh đổi
poke	Giữ luôn một worker slot và kiểm tra lặp lại theo `poke_interval`.	Phản hồi nhanh nhưng chiếm slot suốt thời gian chờ — tốn tài nguyên nếu chờ lâu.
reschedule	Kiểm tra xong thì nhả slot và ngủ tới lần kiểm tra sau.	Tiết kiệm tài nguyên khi chờ lâu; nên dùng cho thời gian chờ dài.

from airflow.sensors.filesystem import FileSensor

wait_file = FileSensor(
    task_id="wait_for_input",
    filepath="/data/input/sales.csv",
    poke_interval=30,            # 30s kiểm tra một lần
    timeout=60 * 60,            # bỏ cuộc sau 1 giờ
    mode="reschedule",          # nhả slot khi chờ
)

XCom — truyền dữ liệu giữa task

Cross-communication

XCom (cross-communication) cho phép một task đẩy một giá trị nhỏ để task khác lấy về. Dùng ti.xcom_push / ti.xcom_pull ở kiểu cổ điển, hoặc đơn giản return giá trị trong TaskFlow (Airflow tự push/pull).

def push_value(ti):
    ti.xcom_push(key="row_count", value=1500)

def pull_value(ti):
    n = ti.xcom_pull(task_ids="producer", key="row_count")
    print(f"Nhận được {n} bản ghi")

producer = PythonOperator(task_id="producer", python_callable=push_value)
consumer = PythonOperator(task_id="consumer", python_callable=pull_value)
producer >> consumer

XCom chỉ cho dữ liệu nhỏ: giá trị XCom lưu trong metadata DB, đừng nhét DataFrame hay file lớn vào đó. Với dữ liệu lớn, truyền đường dẫn (path trên S3/MinIO/HDFS) qua XCom, còn dữ liệu thật để ở storage.

Hooks, Connections & Variables

Quản lý kết nối và cấu hình

Connections lưu thông tin kết nối tới hệ thống ngoài (host, user, password, port của DB/API) — khai báo một lần trong UI (Admin → Connections) hoặc qua biến môi trường, rồi tham chiếu bằng conn_id. Nhờ vậy không hardcode mật khẩu trong code DAG.

Hooks là lớp giao tiếp cấp thấp với hệ thống ngoài, đọc thông tin từ Connection. Lấy connection trong code bằng BaseHook.get_connection:

from airflow.hooks.base import BaseHook

conn = BaseHook.get_connection("my_postgres")
print(conn.host, conn.login, conn.port)   # dùng để mở kết nối

Variables là cặp key–value lưu cấu hình động (đường dẫn, cờ bật/tắt, danh sách) — quản lý trong UI (Admin → Variables) và đọc trong DAG:

from airflow.models import Variable

input_path = Variable.get("input_path", default_var="/data/input")

Best practice: dùng Connection cho thông tin nhạy cảm (mật khẩu DB/API key), dùng Variable cho cấu hình thường — đừng nhúng cứng trong file DAG đã đẩy lên Git.

Branching — rẽ nhánh động

BranchPythonOperator

BranchPythonOperator chạy một hàm trả về task_id (hoặc danh sách task_id) của nhánh sẽ chạy tiếp; các nhánh còn lại bị skip.

from airflow.operators.python import BranchPythonOperator

def choose_branch(ti):
    n = ti.xcom_pull(task_ids="extract", key="row_count")
    return "process_big" if n > 1000 else "process_small"

branch = BranchPythonOperator(
    task_id="branch",
    python_callable=choose_branch,
)
branch >> [process_big, process_small]

Lưu ý downstream: task nằm sau nhánh bị skip cũng có thể bị skip theo. Khi cần một task "gộp" luôn chạy, đặt trigger_rule="none_failed_min_one_success" cho task đó.