Toán tử và XCom

Bạn sẽ học được

• Phân biệt operator truyền thống và TaskFlow API.
• Truyền dữ liệu giữa task đúng cách với XCom.
• Vì sao không đưa DataFrame qua XCom.
• Dùng task group để nhóm logic.
• Chạy song song động với expand.

Khái niệm cốt lõi

Mỗi ô trong DAG là một task, được tạo từ một operator.

TaskFlow API bọc PythonOperator trong decorator @task, cho phép truyền dữ liệu như gọi hàm thường.

Dữ liệu giữa các task

Airflow không truyền dữ liệu lớn giữa task. Thứ được truyền là XCom: một giá trị nhỏ lưu trong metadata database.

task A trả về {"path": "s3://bucket/orders/2026-03-01.parquet"}
       ↓ XCom
task B đọc XCom, mở file theo path

Quy tắc: truyền con trỏ, không truyền dữ liệu.

Ví dụ TaskFlow

from airflow.decorators import dag, task

@dag(...)
def orders_pipeline():
    @task
    def extract(ds=None):
        path = f"/data/orders/{ds}.parquet"
        return {"path": path, "rows": 12000}

    @task
    def transform(meta):
        # đọc từ meta["path"], ghi ra file mới
        out = meta["path"].replace(".parquet", "_clean.parquet")
        return {"path": out, "rows": meta["rows"]}

    @task
    def load(meta):
        print(f"nạp {meta['rows']} dòng từ {meta['path']}")

    load(transform(extract()))

Mỗi giá trị trả về được đẩy vào XCom, task sau nhận qua tham số.

Giới hạn của XCom

XCom mặc định lưu trong metadata DB, có giới hạn kích thước (thường vài chục KB tuỳ cấu hình backend). Đưa DataFrame 500 MB vào XCom sẽ:

• Làm phình metadata DB.
• Chậm toàn bộ hệ thống.
• Có thể thất bại âm thầm hoặc làm hỏng scheduler.

Đúng: ghi dữ liệu ra S3, GCS, hoặc bảng staging; XCom chỉ giữ đường dẫn và vài số liệu tổng hợp.

Task group

Khi DAG có nhiều nhánh, nhóm chúng để dễ đọc:

from airflow.utils.task_group import TaskGroup

with TaskGroup("lam_sach") as clean:
    drop_duplicates()
    fill_nulls()
    normalize_text()

extract() >> clean >> load()

Task group chỉ là cách tổ chức giao diện; bên dưới vẫn là các task riêng biệt.

Chạy song song động

Số vùng dữ liệu có thể thay đổi theo ngày. expand tạo một task cho mỗi phần tử:

@task
def list_regions(ds=None):
    return ["bac", "trung", "nam"]

@task
def process_region(region, ds=None):
    print(f"xử lý {region} ngày {ds}")

process_region.expand(region=list_regions())

Nếu ngày đó có bốn vùng, Airflow tạo bốn task. Đây là dynamic task mapping — không cần biết trước số lượng khi viết DAG.

Giới hạn: max_active_tis_per_dagrun để tránh mở quá nhiều task cùng lúc.

Chọn operator nào

Ưu tiên operator có sẵn thay vì viết Python gọi subprocess — log, retry và trạng thái được xử lý đúng.

Lỗi thường gặp

• Đưa DataFrame qua XCom: phình DB, chậm, có thể hỏng.
• Trả về None rồi task sau cần dữ liệu: lỗi NoneType.
• Nhồi toàn bộ logic vào một @task: mất khả năng chạy lại từng bước.
• Dùng BashOperator cho logic Python: khó test.
• Quên ds trong task: xử lý sai ngày.
• Mở quá nhiều task động: executor quá tải.

Thực hành

1. Viết DAG TaskFlow ba bước truyền đường dẫn qua XCom.
2. Thử truyền một dict lớn và quan sát lỗi.
3. Nhóm ba task làm sạch vào một TaskGroup.
4. Dùng expand để xử lý danh sách vùng.
5. Đổi một BashOperator sang @task và so sánh khả năng test.

Lỗi thường gặp

Tóm tắt

• Task được tạo từ operator; TaskFlow đơn giản hoá Python.
• XCom chỉ dành cho dữ liệu nhỏ; truyền đường dẫn thay vì dữ liệu.
• Task group tổ chức DAG, không thay đổi cách chạy.
• expand tạo task động theo danh sách đầu vào.
• Ưu tiên operator có sẵn khi có thể.