Airflow 8 — Patterns, best practices & testing

Đây là bài khép lại series Airflow chuyên sâu. Sau khi đã đi qua kiến trúc, TaskFlow, scheduling và vận hành, câu hỏi cuối cùng là: viết một DAG như thế nào để nó sống được nhiều năm mà không trở thành gánh nặng? Bài này gom lại những nguyên tắc thiết kế, mẫu tích hợp thực chiến, cách kiểm thử, cách chèn kiểm tra chất lượng dữ liệu, và một danh sách các anti-pattern mà gần như đội nào cũng vấp phải. Xuyên suốt bài, ví dụ minh hoạ là pipeline dữ liệu của một ngân hàng: mỗi ngày ingest giao dịch, transform bằng dbt, kiểm tra chất lượng rồi publish cho báo cáo rủi ro.

Tất cả nội dung theo Airflow 3.x — nơi TaskFlow, Assets và Task SDK đã là công dân hạng nhất chứ không còn là tính năng thử nghiệm.

Nguyên tắc vàng khi viết DAG

Trước khi bàn mẫu thiết kế, phải thấm ba nguyên tắc nền tảng. Vi phạm chúng thì mọi "best practice" khác đều vô nghĩa.

Idempotency — chạy lại cho cùng một kết quả

Một task idempotent là task mà chạy 1 lần hay chạy 5 lần với cùng tham số đều cho ra cùng trạng thái cuối cùng. Đây là nguyên tắc quan trọng nhất, vì Airflow được thiết kế xoay quanh việc retry và backfill: task sẽ chạy lại khi lỗi mạng, khi bạn clear thủ công, khi backfill lại một khoảng ngày.

Vấn đề kinh điển là task ghi kiểu "append": chạy lại một lần là dữ liệu nhân đôi. Cách sửa là gắn việc ghi vào một khoảng thời gian xác định (data interval) và ghi đè thay vì chèn thêm:

• Xoá partition của ngày đang xử lý rồi ghi lại (delete-then-insert theo partition), thay vì INSERT thô.
• Với warehouse: dùng MERGE/INSERT OVERWRITE theo khoá ngày.
• Với file: ghi vào đường dẫn có phân vùng theo ngày .../dt=2026-06-30/, chạy lại chỉ đè đúng phân vùng đó.

Chìa khoá là mỗi task run phải biết nó đang xử lý khoảng nào thông qua các biến ngữ cảnh như data_interval_start / data_interval_end, không phải datetime.now(). Dùng now() là tự tay phá idempotency vì kết quả phụ thuộc vào thời điểm chạy.

Atomicity — một task, một việc rõ ràng

Một task nên làm đúng một đơn vị công việc có thể thành công hoặc thất bại trọn vẹn. Đừng nhồi "tải file + parse + ghi DB + gửi email" vào một task. Khi task đó fail ở bước ghi DB, retry sẽ tải lại file và parse lại từ đầu — lãng phí và dễ tạo hiệu ứng phụ.

Tách nhỏ giúp: retry đúng chỗ lỗi, đọc log rõ ràng, chạy song song được, và tái sử dụng. Nhưng đừng cực đoan chia quá nhỏ tới mức 200 task cho một việc tuyến tính — mỗi task có chi phí scheduling. Quy tắc thực dụng: chia theo ranh giới hệ thống bên ngoài (mỗi lần gọi ra một hệ khác là một task) và theo ranh giới retry (thứ gì cần retry riêng thì tách riêng).

Airflow điều phối, không xử lý dữ liệu lớn

Đây là hiểu lầm tốn kém nhất của người mới. Airflow là orchestrator, không phải engine xử lý dữ liệu. Worker của Airflow không nên đọc 50 triệu dòng vào pandas rồi join trong bộ nhớ. Việc nặng phải đẩy sang hệ ngoài: Spark, warehouse (BigQuery/Snowflake), dbt. Airflow chỉ gửi lệnh "chạy đi" và chờ kết quả.

Hệ quả trực tiếp: không truyền dữ liệu lớn qua XCom. XCom được thiết kế cho metadata nhỏ (đường dẫn file, số dòng, id job) — mặc định lưu trong metadata database. Đẩy một DataFrame qua XCom sẽ phình database và làm chậm scheduler. Nếu thật sự cần chuyển dữ liệu lớn giữa task, hãy ghi ra object storage rồi truyền đường dẫn qua XCom, hoặc cấu hình custom XCom backend trỏ về S3/GCS. (Xem chi tiết cơ chế XCom ở bài TaskFlow & XCom.)

DAG phải parse nhanh

File DAG được scheduler đọc lại liên tục (dag-processor định kỳ quét). Vì vậy code ở top-level không được làm việc nặng: không gọi API, không query DB, không đọc file lớn, không Variable.get() ở cấp module. Mọi lời gọi như vậy sẽ chạy mỗi lần parse — có thể vài giây một lần — làm nghẽn scheduler và trì hoãn mọi DAG khác.

Nguyên tắc: top-level chỉ định nghĩa cấu trúc. Mọi I/O phải nằm trong hàm task, chỉ chạy khi task thực thi.

Các mẫu thiết kế thường gặp

ETL vs ELT

• ETL (Extract → Transform → Load): biến đổi dữ liệu trước khi nạp vào đích. Phù hợp khi transform phải làm ngoài warehouse (ví dụ parse file nhị phân bằng Spark) hoặc khi cần làm sạch dữ liệu nhạy cảm trước khi lưu.
• ELT (Extract → Load → Transform): nạp dữ liệu thô vào warehouse trước, rồi transform bằng SQL ngay trong warehouse (thường qua dbt). Đây là mẫu chủ đạo hiện nay vì warehouse hiện đại rẻ và mạnh, đồng thời giữ được dữ liệu thô để tái xử lý.

Với Airflow, ELT thường sạch hơn: task ingest chỉ đổ dữ liệu thô, task transform chỉ trigger dbt. Airflow không đụng vào dữ liệu.

Orchestrate Spark job

Có hai cách phổ biến để chạy Spark từ Airflow, và cả hai đều tôn trọng nguyên tắc "đẩy compute ra ngoài":

• SparkSubmitOperator: submit job tới một Spark cluster/YARN có sẵn. Đơn giản khi đã có hạ tầng Spark.
• KubernetesPodOperator: đóng gói job Spark (hoặc bất kỳ workload nào) vào một Pod chạy trên Kubernetes. Cách này cô lập dependency tốt, dễ scale, và là mẫu được ưa chuộng trong môi trường cloud-native. Airflow chỉ tạo Pod, theo dõi, rồi thu log.

Điểm chung: Airflow submit và chờ (tốt nhất là dạng deferrable để không giữ worker slot), job thật chạy nơi khác.

Trigger dbt

dbt là công cụ transform SQL trong warehouse — cặp bài trùng với Airflow trong mẫu ELT (xem bài dbt trong series Data Engineering). Có hai cách tích hợp:

• BashOperator gọi dbt run/dbt test: đơn giản, nhưng cả project dbt là một task khối, khó nhìn được model nào fail.
• Astronomer Cosmos: parse manifest.json của dbt và tự sinh ra một task Airflow cho mỗi model/test. Nhờ đó bạn thấy đồ thị phụ thuộc dbt ngay trong UI Airflow, retry được từng model, và ghép kiểm tra chất lượng vào giữa. Đây là mẫu được khuyến nghị cho project dbt lớn.

Kafka: tiêu thụ và đổ dữ liệu

Airflow không phải công cụ streaming — nó theo batch. Với Kafka, mẫu điển hình là micro-batch: một DAG chạy định kỳ, tiêu thụ các message tích luỹ từ topic, ghi thành file/bảng. Airflow cung cấp provider Kafka với các operator/hook để consume và produce. Còn xử lý stream thật sự (millisecond latency) thì để cho Kafka Streams/Flink; Airflow chỉ điều phối các job batch quanh Kafka.

Fan-out bằng Dynamic Task Mapping

Khi cần chạy cùng một logic trên một danh sách phần tử chưa biết trước số lượng (ví dụ: xử lý N file vừa xuất hiện, N chi nhánh), dùng Dynamic Task Mapping (.expand()). Airflow sẽ sinh ra một task mapped cho mỗi phần tử tại runtime. Đây là cách đúng đắn thay cho việc dùng vòng lặp Python ở top-level để tạo task tĩnh (làm phình DAG và cần sửa code mỗi khi số lượng đổi).

Nối pipeline bằng Assets thay vì sensor

Mẫu cũ: DAG B dùng sensor chờ DAG A xong. Sensor poke liên tục, tốn slot, và ràng buộc thời gian ngầm. Airflow 3 đẩy mạnh Assets (data-aware scheduling): DAG A khai báo nó produce một Asset; DAG B khai báo nó chạy khi Asset đó được cập nhật. Không còn sensor, không còn đoán lịch — pipeline nối theo dữ liệu, đúng bản chất. (Chi tiết ở bài Scheduling & Assets.)

Branching và trigger rules

• @task.branch: chọn nhánh chạy dựa trên điều kiện runtime (ví dụ: nếu có dữ liệu thì transform, không thì skip). Task không được chọn sẽ bị skip.
• Trigger rules: mặc định một task chạy khi mọi upstream success. Đôi khi cần khác: all_done (chạy dù upstream fail — hợp cho task dọn dẹp/thông báo), none_failed_min_one_success (hợp sau branch). Chọn đúng trigger rule tránh việc task cleanup bị skip oan khi có nhánh không chạy.

Sơ đồ pipeline ELT theo Asset

Điểm mấu chốt của sơ đồ: DAG_B không chờ theo giờ, mà chạy khi Asset raw.transactions được cập nhật; và kiểm tra chất lượng nằm trước bước publish để fail sớm.

Kiểm thử DAG

DAG là code, nên phải test như code. Có ba tầng kiểm thử, từ rẻ đến đắt.

Tầng 1: kiểm tra tính hợp lệ (validation test)

Test rẻ nhất và bắt buộc: đảm bảo mọi file DAG import được (không lỗi cú pháp, không lỗi top-level), không có chu trình, và tuân theo quy ước nội bộ (mọi DAG phải có owner, có tags, có retries...). Test này bắt được lỗi phổ biến nhất: DAG hỏng do sửa code mà không chạy thử.

# test_dag_integrity.py — (minh hoạ)
from airflow.models import DagBag


def test_no_import_errors():
    dagbag = DagBag(include_examples=False)
    assert not dagbag.import_errors, f"Lỗi import DAG: {dagbag.import_errors}"


def test_dags_have_required_defaults():
    dagbag = DagBag(include_examples=False)
    for dag_id, dag in dagbag.dags.items():
        assert dag.tags, f"{dag_id} thiếu tags"
        assert dag.default_args.get("retries", 0) >= 1, f"{dag_id} nên có retries"

Tầng 2: unit test cho logic Python

Đây là lý do quan trọng để tách logic nghiệp vụ ra khỏi Airflow. Nếu bên trong @task bạn viết logic thẳng vào đó, bạn không test được nếu không dựng cả Airflow. Mẫu đúng: hàm thuần Python nằm ngoài, @task chỉ gọi nó.

# transforms.py — logic thuần, không phụ thuộc Airflow (minh hoạ)
def normalize_amount(raw: str) -> float:
    return round(float(raw.replace(",", "")), 2)


# test_transforms.py — (minh hoạ)
from transforms import normalize_amount


def test_normalize_amount():
    assert normalize_amount("1,234.5") == 1234.5
    assert normalize_amount("0") == 0.0

Trong DAG, @task chỉ là lớp vỏ mỏng gọi normalize_amount. Logic được test độc lập, chạy trong mili-giây, không cần database.

Tầng 3: chạy thử DAG với dag.test()

Airflow cung cấp dag.test() để chạy toàn bộ DAG trong một tiến trình duy nhất, cục bộ, không cần scheduler/worker. Rất tiện để debug end-to-end trên máy dev, và có thể đưa vào CI cho các DAG nhỏ (nên mock các hệ ngoài).

# chạy: python dags/ingest_giao_dich.py  — (minh hoạ)
if __name__ == "__main__":
    from ingest_giao_dich import dag
    dag.test()

Trong CI, tối thiểu nên chạy tầng 1 (integrity) trên mọi PR; tầng 2 cho các module logic; tầng 3 chọn lọc.

Chất lượng dữ liệu như một bước trong pipeline

Một pipeline "chạy xong" không có nghĩa dữ liệu "đúng". Best practice là chèn kiểm tra chất lượng thành một task tường minh, đặt giữa transform và publish, và fail sớm nếu dữ liệu sai — thà dừng còn hơn publish số liệu rác cho báo cáo rủi ro.

Các cách phổ biến:

• SQL check operators: Airflow có sẵn các operator như SQLColumnCheckOperator, SQLTableCheckOperator để chạy assertion trực tiếp trên warehouse: cột không null, giá trị trong khoảng, số dòng > 0, tổng khớp. Nhẹ, không thêm phụ thuộc.
• Great Expectations: framework kiểm thử dữ liệu với "expectation suite" phong phú, sinh cả tài liệu dữ liệu. Mạnh nhưng nặng hơn.
• Soda: khai báo check bằng YAML (SodaCL), chạy như một task; thiên về giám sát chất lượng liên tục.

Nguyên tắc thiết kế: check là gatekeeper. Task publish phụ thuộc task check; check fail thì publish không chạy, và một task alert (trigger rule one_failed) gửi cảnh báo. Đừng để check chỉ ghi log rồi vẫn publish.

Các anti-pattern cần tránh

Danh sách này gói lại những sai lầm gặp đi gặp lại:

• Top-level code nặng: gọi API/DB, đọc file, Variable.get(), Connection.get() ở cấp module. Làm scheduler chậm và DAG parse lỗi. Đưa hết vào trong task.
• DAG khổng lồ: một DAG với hàng trăm task không liên quan. Khó đọc, khó retry, chậm parse. Tách thành nhiều DAG nối bằng Asset.
• Phụ thuộc thời gian ngầm giữa các DAG: "DAG B đặt chạy sau B 30 phút vì đoán A xong lúc đó". Cực kỳ giòn. Dùng Assets/data-aware scheduling.
• Dùng Variable trong top-level: mỗi lần parse là một query vào metadata DB. Nếu cần config động, đọc trong task, hoặc dùng Variable.get với template ở render time.
• Quá nhiều sensor poke: hàng chục sensor mode poke giữ chặt worker slot. Chuyển sang reschedule mode, deferrable sensor, hoặc bỏ hẳn nhờ Assets.
• Hardcode secret: mật khẩu/token nhét thẳng trong code hoặc Variable thường. Dùng Connections và secrets backend (Vault, AWS/GCP Secret Manager). (Xem bài Vận hành production.)
• Task không idempotent: append thô, dùng now(), ghi không theo partition. Retry/backfill sẽ nhân đôi hoặc làm hỏng dữ liệu.
• Xử lý dữ liệu lớn trong worker / XCom to: đã bàn — đẩy compute ra ngoài, XCom chỉ chứa metadata.

Điểm mới của Airflow 3 nên tận dụng

Airflow 3 không chỉ là bản nâng cấp — nó thay đổi cách viết pipeline theo hướng bền vững hơn:

• TaskFlow + Task SDK: viết pipeline bằng hàm Python decorator (@task, @dag) thay vì lắp operator thủ công; dependency và truyền XCom suy ra từ lời gọi hàm. Task SDK tách rõ lớp thực thi task khỏi core, giúp mã task ổn định và cô lập hơn.
• Assets / event-driven scheduling: nối pipeline theo dữ liệu, giảm sensor. Là hướng thiết kế được khuyến khích cho pipeline nhiều tầng.
• DAG versioning: Airflow 3 theo dõi phiên bản của DAG; khi xem một lần chạy cũ, UI hiển thị đúng cấu trúc DAG tại thời điểm đó — hết cảnh nhìn run cũ bằng code mới, đỡ rối khi debug.
• Deferrable operators: task "chờ" (chờ file, chờ job Spark, chờ giờ) nhả worker slot và giao việc chờ cho triggerer. Tiết kiệm tài nguyên đáng kể so với sensor poke.
• Backfill qua UI/API: backfill được quản lý bởi scheduler và điều khiển qua UI/API, thay vì lệnh CLI rời rạc — dễ theo dõi và kiểm soát hơn.

Use case thực tế: refactor một DAG xấu thành tốt

Bắt đầu với một DAG "đời thực" đầy vấn đề: pipeline giao dịch ngân hàng.

# TRƯỚC — nhiều anti-pattern (minh hoạ)
import pandas as pd
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.models import Variable
import datetime

# top-level: query DB + đọc Variable mỗi lần parse (xấu)
cfg = Variable.get("txn_config", deserialize_json=True)
DB_URL = "postgres://user:secret@host/db"  # hardcode secret (xấu)


def do_everything():
    df = pd.read_sql("SELECT * FROM transactions", DB_URL)  # kéo hết vào worker
    df["amount"] = df["amount"].str.replace(",", "").astype(float)
    df.to_sql("report", DB_URL, if_exists="append")  # append -> không idempotent
    # gửi email... tất cả trong một task (không atomic)


with DAG("txn", schedule="@daily", start_date=datetime.datetime(2024, 1, 1)):
    PythonOperator(task_id="run", python_callable=do_everything)

Các vấn đề: secret hardcode, Variable.get() + query top-level, một task làm mọi việc, kéo toàn bộ bảng vào worker, ghi append (không idempotent), không retry, không kiểm tra chất lượng.

Bản refactor tôn trọng nguyên tắc — Airflow chỉ điều phối, logic tách riêng, idempotent theo ngày, có check chất lượng:

# SAU — theo best practice (minh hoạ)
from __future__ import annotations
import pendulum
from airflow.sdk import dag, task, Asset
from airflow.providers.common.sql.operators.sql import SQLTableCheckOperator
from transforms import normalize_amount  # logic thuần, test riêng

raw_txn = Asset("warehouse://raw/transactions")
report = Asset("warehouse://mart/txn_report")

default_args = {"owner": "data-platform", "retries": 2,
                "retry_delay": pendulum.duration(minutes=5)}


@dag(schedule="@daily", start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
     catchup=False, default_args=default_args, tags=["banking", "elt"])
def txn_report():

    @task
    def extract(data_interval_start=None, data_interval_end=None) -> str:
        # dùng hook + connection (không hardcode secret); trả về đường dẫn, KHÔNG trả data
        path = f"s3://raw/txn/dt={data_interval_start.date()}/"
        # ... đọc theo khoảng [start, end) rồi ghi ra path (idempotent theo ngày)
        return path

    @task
    def load(path: str, data_interval_start=None) -> None:
        dt = data_interval_start.date()
        # MERGE / INSERT OVERWRITE theo partition dt -> chạy lại đè đúng ngày
        ...

    # dbt transform ở đây thường dùng Cosmos; rút gọn thành 1 task minh hoạ
    @task(outlets=[report])
    def transform() -> None:
        ...  # trigger dbt (warehouse tự transform)

    dq_check = SQLTableCheckOperator(
        task_id="dq_check",
        table="mart.txn_report",
        checks={"row_count": {"check_statement": "COUNT(*) > 0"},
                "amount_non_negative": {"check_statement": "MIN(amount) >= 0"}},
    )

    p = extract()
    load(p) >> transform() >> dq_check   # check là gatekeeper trước khi coi là xong


txn_report()

Khác biệt: không secret trong code (dùng connection/hook), không I/O top-level, mỗi task một việc, ghi idempotent theo partition ngày, XCom chỉ chứa đường dẫn, có retries, có kiểm tra chất lượng chặn ở cuối, và DAG được kích bằng dữ liệu qua Asset thay vì đoán lịch. Logic normalize_amount nằm ở module riêng nên unit test được.

Ghi nhớ

• Idempotency là nguyên tắc số một: gắn công việc vào data interval, ghi đè theo partition, không dùng now().
• Airflow điều phối, không xử lý dữ liệu lớn. Đẩy compute sang Spark/warehouse/dbt; XCom chỉ chứa metadata.
• Tách logic Python thuần ra khỏi @task để unit test được; kiểm tra DAG import trong CI.
• Chèn kiểm tra chất lượng như một task gatekeeper, fail sớm trước khi publish.
• Nối pipeline bằng Assets, không bằng sensor chờ hay lịch đoán mò.

Checklist review một DAG trước khi lên production

• Top-level không có I/O (không API/DB/đọc file/Variable.get()/Connection.get()).
• Mọi task idempotent: ghi đè theo partition/MERGE, không append thô, không now().
• Mỗi task atomic, chia theo ranh giới hệ ngoài và ranh giới retry.
• Không xử lý dữ liệu lớn trong worker; compute nặng đẩy ra Spark/warehouse/dbt.
• XCom chỉ chứa metadata nhỏ; dữ liệu lớn đi qua object storage.
• Có retries + retry_delay hợp lý; task lỗi tạm thời không làm sập cả pipeline.
• Secret qua Connections/secrets backend, không hardcode, không nằm trong Variable thường.
• Có bước kiểm tra chất lượng dữ liệu chặn trước publish; có alert khi check fail.
• catchup được đặt có chủ đích; start_date cố định, không dùng ngày động.
• Có owner, tags, mô tả rõ; đặt lịch bằng Asset khi phụ thuộc dữ liệu.
• DAG parse nhanh; đã qua integrity test (import OK, không cycle) trong CI.
• Logic nghiệp vụ tách ra module riêng và có unit test.
• Task "chờ" dùng deferrable/reschedule, không phải sensor poke ôm slot.

Đến đây khép lại series Airflow chuyên sâu: từ kiến trúc, TaskFlow & XCom, scheduling & Assets, vận hành production, đến những mẫu thiết kế và kỷ luật kỹ thuật giúp DAG của bạn bền vững. Một pipeline tốt không phải pipeline phức tạp nhất — mà là pipeline mà nửa năm sau bạn (hoặc người kế nhiệm) vẫn hiểu, sửa và tin tưởng được.