Airflow 4 — TaskFlow, XCom, templating & Dynamic Task Mapping

Ở bài về DAG & Task chúng ta đã thấy cách khai báo pipeline và dựng dependency. Ở bài Operators & Hooks là cách "làm việc" với hệ thống bên ngoài. Bài này trả lời một câu hỏi rất thực tế mà ai làm Airflow cũng gặp: làm sao để một task đưa kết quả cho task tiếp theo?

Câu trả lời của Airflow là XCom (cross-communication), và cách viết code hiện đại nhất để dùng nó là TaskFlow API. Chúng ta cũng sẽ đi qua Jinja templating, params, và cuối cùng là Dynamic Task Mapping — cơ chế map-reduce để sinh nhiều task instance song song theo dữ liệu runtime. Nội dung dựa trên Airflow 3.x.

1. Vấn đề: task không chia sẻ bộ nhớ

Mỗi task instance trong Airflow chạy trong một tiến trình (thậm chí một máy/pod) độc lập. Không có biến toàn cục, không có bộ nhớ dùng chung. Task B không thấy biến Python mà task A vừa tạo ra:

def task_a():
    result = expensive_query()   # (minh hoạ)
    # result này biến mất khi task_a kết thúc

def task_b():
    print(result)   # NameError — task_b không hề biết 'result'

Airflow giải quyết bằng cách cho task push một giá trị vào metadata DB dưới một key, và task sau pull giá trị đó ra. Đó chính là XCom.

2. XCom — cross-communication

XCom là một cơ chế lưu các mẩu dữ liệu nhỏ vào bảng xcom trong metadata database, gắn với (dag_id, run_id, task_id, key, map_index). Task sau đọc lại theo cùng khoá.

2.1 Cách cổ điển: xcom_push / xcom_pull

Trước TaskFlow, ta thao tác XCom thủ công qua đối tượng task instance (ti) trong context:

from airflow.sdk import DAG, task  # (minh hoạ, Airflow 3.x)
from airflow.providers.standard.operators.python import PythonOperator

def push_fn(**context):
    ti = context["ti"]
    ti.xcom_push(key="row_count", value=42)

def pull_fn(**context):
    ti = context["ti"]
    n = ti.xcom_pull(task_ids="push", key="row_count")
    print(f"Nhận được {n} dòng")

# return của PythonOperator tự động đẩy vào XCom key="return_value"

Cách này hoạt động tốt nhưng nhiều boilerplate: phải nhớ task_ids, key, phải khai báo dependency push >> pull bằng tay, và dễ sai chính tả tên key.

2.2 XCom lưu ở đâu, và giới hạn kích thước

Mặc định XCom được serialize và lưu thẳng trong metadata DB (Postgres/MySQL). Đây là điểm quan trọng nhất cần nhớ:

• Cột lưu giá trị có giới hạn kích thước phụ thuộc backend DB (ví dụ Postgres dùng kiểu bytea/JSON, MySQL có giới hạn LONGBLOB nhưng vẫn không nên nhét dữ liệu lớn).
• XCom được thiết kế cho metadata nhỏ: số dòng, đường dẫn file, id, cờ trạng thái, một dict cấu hình ngắn.
• KHÔNG đẩy DataFrame lớn, danh sách hàng nghìn bản ghi, hay file nội dung qua XCom.

Rủi ro khi lạm dụng: metadata DB phình to nhanh, mỗi lần scheduler/worker serialize–deserialize tốn CPU và băng thông, DB trở thành nút thắt cổ chai, và có thể vượt giới hạn cột gây lỗi task. Metadata DB là "trái tim" của cluster — làm nó chậm là làm cả Airflow chậm.

2.3 Custom XCom backend & Object Storage (Airflow 3)

Giải pháp cho payload lớn: giữ dữ liệu thật ở object storage (S3/GCS/Azure Blob), chỉ lưu một tham chiếu (URI) trong DB.

Airflow cho phép override cơ chế serialize bằng cách kế thừa BaseXCom và ghi đè serialize_value / deserialize_value:

# custom_xcom_backend.py — (minh hoạ, rút gọn)
import uuid, pandas as pd
from airflow.models.xcom import BaseXCom
from airflow.providers.amazon.aws.hooks.s3 import S3Hook

BUCKET = "my-airflow-xcom"

class S3XComBackend(BaseXCom):
    @staticmethod
    def serialize_value(value, **kwargs):
        # Chỉ đẩy DataFrame lớn lên S3, còn lại giữ nguyên trong DB
        if isinstance(value, pd.DataFrame):
            key = f"xcom/{uuid.uuid4()}.parquet"
            hook = S3Hook()
            local = f"/tmp/{uuid.uuid4()}.parquet"
            value.to_parquet(local)
            hook.load_file(local, key=key, bucket_name=BUCKET, replace=True)
            value = f"s3://{BUCKET}/{key}"   # chỉ URI được lưu trong DB
        return BaseXCom.serialize_value(value)

    @staticmethod
    def deserialize_value(result):
        value = BaseXCom.deserialize_value(result)
        if isinstance(value, str) and value.startswith("s3://"):
            hook = S3Hook()
            _, _, rest = value.partition("s3://")
            bucket, _, key = rest.partition("/")
            path = hook.download_file(key=key, bucket_name=bucket)
            return pd.read_parquet(path)
        return value

Kích hoạt bằng cấu hình [core] xcom_backend = custom_xcom_backend.S3XComBackend.

Airflow 3 làm việc này dễ hơn nhiều với Object Storage XCom Backend dựng sẵn trong provider common.io. Bạn chỉ cần cấu hình, không cần tự viết class:

[core]
xcom_backend = airflow.providers.common.io.xcom.backend.XComObjectStorageBackend

[common.io]
xcom_objectstorage_path = s3://my-conn@my-bucket/xcom
xcom_objectstorage_threshold = 1048576     # > ngưỡng (byte) mới đẩy ra object store
xcom_objectstorage_compression = zip       # nén trước khi lưu (tùy chọn)

Điểm hay: đây là hybrid — object nhỏ hơn threshold vẫn nằm trong DB (nhanh), chỉ object lớn mới ra object storage và DB chỉ giữ tham chiếu. threshold phải > -1 mới bật cơ chế. Nhờ vậy ta được sự tiện lợi của XCom mà tránh làm phình metadata DB.

3. TaskFlow API — viết XCom một cách sạch sẽ

TaskFlow (@task) biến một hàm Python thành task. Điểm mấu chốt: giá trị return tự động được push vào XCom, và khi bạn truyền output đó làm tham số cho hàm task khác, Airflow tự pull XCom và tự dựng dependency giúp bạn.

from airflow.sdk import dag, task   # (minh hoạ, Airflow 3.x)
import pendulum

@dag(
    schedule="@daily",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
    tags=["taskflow-demo"],
)
def etl_pipeline():

    @task
    def extract() -> dict:
        # return tự đẩy vào XCom (key="return_value")
        return {"order_ids": [1, 2, 3, 4, 5]}

    @task
    def transform(data: dict) -> int:
        # 'data' được Airflow tự pull từ XCom của extract()
        return len(data["order_ids"])

    @task
    def load(count: int) -> None:
        print(f"Đã xử lý {count} đơn hàng")

    orders = extract()
    n = transform(orders)     # dependency extract -> transform tự sinh
    load(n)                   # dependency transform -> load tự sinh

etl_pipeline()

So sánh với cách cổ điển: không còn xcom_push, không xcom_pull, không phải viết >> bằng tay, không nhớ tên key. Airflow suy ra dependency từ chính luồng dữ liệu giữa các hàm. Code đọc gần như code Python bình thường.

Vài lưu ý về TaskFlow:

• Bên dưới, @task vẫn tạo ra một operator (_PythonDecoratedOperator) và vẫn dùng XCom — nên mọi giới hạn kích thước XCom vẫn áp dụng. TaskFlow không phải phép màu bỏ qua giới hạn.
• Có thể trộn TaskFlow với operator cổ điển: output của @task có thể truyền vào operator, và ngược lại lấy operator.output để đưa vào @task.
• multiple_outputs=True (hoặc return type là dict) cho phép tách một dict thành nhiều XCom key riêng, giúp downstream chọn đúng phần cần.

4. Jinja templating

Nhiều trường của operator không nhận giá trị tĩnh mà nhận template Jinja, được render lại tại thời điểm chạy cho từng DAG run. Các trường này liệt kê trong template_fields của operator (ví dụ bash_command của BashOperator, sql của các SQL operator).

from airflow.providers.standard.operators.bash import BashOperator  # (minh hoạ)

run = BashOperator(
    task_id="run_report",
    bash_command="python report.py --date {{ ds }} --run {{ run_id }}",
)

4.1 Bảng biến template hay dùng (Airflow 3.x)

Quan trọng cho ai lên từ Airflow 2: execution_date đã bị bỏ. Trong Airflow 3 hãy dùng logical_date, ds, hoặc — chuẩn nhất cho pipeline theo khoảng thời gian — data_interval_start / data_interval_end. Ba biến interval này mô tả rõ ràng "run này xử lý dữ liệu của khoảng nào", tránh nhầm lẫn giữa "thời điểm chạy" và "thời điểm dữ liệu".

Lưu ý thêm: DAG được kích hoạt bởi Asset (asset-triggered) không có logical date, nên các biến thời gian như logical_date, ds, data_interval_* sẽ không có sẵn — cần thiết kế pipeline không phụ thuộc chúng.

4.2 Vì sao dùng template thay vì datetime.now()

Nếu bạn hardcode datetime.now() trong DAG file, giá trị được tính khi scheduler parse file, không phải khi task chạy, và mất hoàn toàn ý nghĩa khi backfill. Template Jinja được render đúng cho từng run — kể cả khi chạy lại một ngày trong quá khứ — nên là cách duy nhất đúng để lấy "ngày của run này".

5. Params — tham số hoá DAG

params cho phép định nghĩa tham số ở cấp DAG, có giá trị mặc định, và có thể ghi đè khi trigger thủ công (UI hiện form nhập). Airflow hỗ trợ validation qua schema kiểu JSON Schema.

from airflow.sdk import dag, task
from airflow.models.param import Param   # (minh hoạ)
import pendulum

@dag(
    schedule=None,
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    params={
        "country": Param("VN", type="string", enum=["VN", "US", "JP"]),
        "limit": Param(100, type="integer", minimum=1, maximum=10000),
    },
)
def parametrized():

    @task
    def run(**context):
        p = context["params"]
        print(f"country={p['country']}, limit={p['limit']}")

    run()

parametrized()

Trong template dùng {{ params.country }}. Nếu người dùng nhập limit=99999 vượt maximum, Airflow từ chối trigger — validation giúp chặn cấu hình sai ngay từ đầu thay vì để task chạy rồi lỗi giữa chừng.

6. Dynamic Task Mapping — map-reduce trong Airflow

Đôi khi bạn không biết trước số task cần chạy: số file trong thư mục, số partition, số quốc gia trả về từ một API. Dynamic Task Mapping cho phép sinh nhiều mapped task instance song song từ một danh sách dữ liệu runtime — không cần biết số lượng khi viết DAG.

6.1 .expand() và .partial()

• .expand(arg=[...]): với mỗi phần tử trong danh sách, tạo một task instance riêng (map). Danh sách này có thể là output XCom của task trước.
• .partial(arg=value): cố định các tham số giống nhau cho mọi mapped instance.

from airflow.sdk import dag, task   # (minh hoạ, Airflow 3.x)
import pendulum

@dag(schedule="@daily", start_date=pendulum.datetime(2026, 1, 1, tz="UTC"), catchup=False)
def dynamic_demo():

    @task
    def list_files() -> list[str]:
        # runtime mới biết có bao nhiêu file
        return ["a.csv", "b.csv", "c.csv"]

    @task
    def process(file_name: str, env: str) -> int:
        print(f"[{env}] xử lý {file_name}")
        return 1

    @task
    def summarize(results: list[int]) -> None:
        # 'results' là danh sách kết quả của MỌI mapped instance (reduce)
        print(f"Tổng số file xử lý: {sum(results)}")

    files = list_files()
    # env cố định (partial), file_name map theo từng phần tử (expand)
    counts = process.partial(env="prod").expand(file_name=files)
    summarize(counts)

dynamic_demo()

Điểm mấu chốt: process xuất hiện một lần trong code nhưng runtime sẽ có 3 (hoặc N) instance chạy song song, mỗi instance nhận một file_name. Kết quả của tất cả instance gom lại thành một danh sách và đưa vào summarize — đúng mô hình map → reduce.

6.2 expand_kwargs và map trên nhiều chiều

• .expand_kwargs(list_of_dicts): mỗi phần tử là một dict tham số cho một instance — hữu ích khi mỗi map cần nhiều tham số khác nhau, không chỉ một.
• Nếu .expand() nhiều tham số cùng lúc, Airflow tạo tích Descartes (cross product) các giá trị — cẩn thận vì số instance có thể bùng nổ.

6.3 Giới hạn số map

Fan-out không kiểm soát có thể tạo hàng nghìn task và làm nghẽn scheduler/worker. Kiểm soát bằng:

• max_map_length (mặc định 1024): chặn cứng số instance tối đa sinh ra từ một map, tránh sự cố dữ liệu đầu vào bất thường tạo ra quá nhiều task.
• max_active_tis_per_dag / pool: giới hạn đồng thời — có thể có 500 mapped instance nhưng chỉ cho phép 20 chạy cùng lúc để không làm sập hệ thống downstream.

7. Use case thực tế — xử lý N file song song

Bài toán rất phổ biến: mỗi sáng có một số file được đẩy lên S3 (số lượng thay đổi hằng ngày), cần xử lý từng file rồi tổng hợp.

from airflow.sdk import dag, task           # (minh hoạ, Airflow 3.x)
from airflow.providers.amazon.aws.hooks.s3 import S3Hook
import pendulum

@dag(schedule="@daily", start_date=pendulum.datetime(2026, 1, 1, tz="UTC"), catchup=False)
def process_daily_files():

    @task
    def discover(prefix: str, **context) -> list[str]:
        # dùng data_interval_start để lấy đúng ngày của run
        ds = context["data_interval_start"].format("YYYY/MM/DD")
        hook = S3Hook(aws_conn_id="aws_default")
        keys = hook.list_keys(bucket_name="landing", prefix=f"{prefix}/{ds}/")
        return keys or []

    @task(max_active_tis_per_dag=10)   # tối đa 10 instance chạy đồng thời
    def transform_one(key: str) -> dict:
        hook = S3Hook(aws_conn_id="aws_default")
        obj = hook.read_key(key=key, bucket_name="landing")
        rows = len(obj.splitlines())
        # chỉ return metadata NHỎ, không return nội dung file
        return {"key": key, "rows": rows}

    @task
    def report(results: list[dict]) -> None:
        total = sum(r["rows"] for r in results)
        print(f"Đã xử lý {len(results)} file, tổng {total} dòng")

    keys = discover(prefix="orders")
    stats = transform_one.expand(key=keys)   # fan-out theo số file runtime
    report(stats)

process_daily_files()

Vì sao thiết kế này tốt: số file bao nhiêu cũng chạy được mà không sửa DAG; các file xử lý song song (bị chặn ở 10 để bảo vệ hệ thống); và mỗi task chỉ đẩy metadata nhỏ qua XCom ({key, rows}), còn nội dung file luôn ở S3 — đúng nguyên tắc XCom.

8. Khi nào dùng XCom, khi nào ghi ra storage

Quy tắc quyết định gọn:

• Dùng XCom trực tiếp khi dữ liệu nhỏ và là metadata: số đếm, id, đường dẫn, cờ, một dict cấu hình ngắn (kilobyte trở xuống).
• Ghi ra storage (S3/GCS/DB/lakehouse), chỉ truyền URI qua XCom khi dữ liệu là DataFrame, tập bản ghi lớn, hay file. Task sau đọc lại từ storage bằng URI đó.
• Dùng Object Storage XCom Backend / custom backend khi bạn muốn giữ trải nghiệm "return là xong" của TaskFlow nhưng payload thỉnh thoảng lớn — để framework tự động đẩy phần lớn ra object store.

Anti-pattern cần tránh: đẩy DataFrame hàng trăm MB, dump JSON khổng lồ, hay nội dung file thô qua XCom. Nó làm phình metadata DB, chậm scheduler, và có thể vượt giới hạn cột gây fail task. Nhớ: XCom để trao đổi thông tin về dữ liệu, không phải để trao đổi chính dữ liệu lớn.

9. Ghi nhớ

• Task không chia sẻ bộ nhớ; XCom là kênh chính thức để một task chuyển giá trị cho task sau, lưu trong metadata DB.
• TaskFlow (@task): return tự push XCom, truyền output làm tham số tự pull XCom và tự dựng dependency → code sạch, ít boilerplate. Nhưng vẫn chịu giới hạn XCom.
• XCom có giới hạn kích thước; không đẩy DataFrame/file lớn. Dùng Object Storage XCom Backend (Airflow 3, hybrid theo threshold) hoặc custom backend để chỉ giữ tham chiếu trong DB.
• Templating Jinja render tại runtime cho template_fields. Airflow 3 bỏ execution_date — dùng logical_date, ds, và chuẩn nhất là data_interval_start / data_interval_end. Đừng dùng datetime.now() trong DAG file.
• Params tham số hoá DAG, ghi đè khi trigger thủ công, có validation kiểu JSON Schema.
• Dynamic Task Mapping (.expand / .partial / .expand_kwargs) sinh N task instance song song theo dữ liệu runtime — map-reduce của Airflow. Kiểm soát bằng max_map_length và giới hạn đồng thời.
• Nguyên tắc vàng: XCom cho metadata nhỏ, storage cho dữ liệu lớn.

Xem thêm: DAG & Task · Operators & Hooks · Best practices.