Airflow 2 — DAG, Task & cách khai báo pipeline

Nếu bài đầu tiên của series trả lời câu hỏi "Airflow là gì và các thành phần vận hành ra sao", thì bài này trả lời câu hỏi mà bạn sẽ gặp mỗi ngày khi làm việc thật: làm sao viết một pipeline? Đơn vị công việc trong Airflow là DAG, và mỗi DAG được cấu thành từ các Task. Hiểu thấu hai khái niệm này — cùng với cách chúng phụ thuộc lẫn nhau, chạy theo lịch nào, và xử lý lỗi ra sao — là điều kiện cần để viết pipeline chạy được trong môi trường production.

Toàn bộ ví dụ và khái niệm trong bài dựa trên Apache Airflow 3.x, phiên bản có nhiều thay đổi quan trọng so với 2.x mà chúng ta sẽ chỉ rõ khi cần.

DAG là gì?

DAG viết tắt của Directed Acyclic Graph — đồ thị có hướng, không chu trình. Dịch nôm na: một tập hợp các công việc được nối với nhau bằng các mũi tên chỉ chiều, và bạn không bao giờ đi lòng vòng quay về điểm xuất phát.

• Directed (có hướng): mỗi cạnh có chiều — A chạy trước, rồi mới đến B. Airflow luôn biết cái gì phải xong trước cái gì.
• Acyclic (không chu trình): không có vòng lặp. Nếu A → B → C → A thì scheduler sẽ không bao giờ biết bắt đầu từ đâu; Airflow cấm điều này và sẽ báo lỗi khi parse DAG.
• Graph (đồ thị): mỗi node là một Task, mỗi cạnh là một quan hệ phụ thuộc (dependency).

Điều quan trọng cần nhớ: DAG chỉ mô tả cấu trúc và lịch chạy, nó không tự làm việc. Việc thực thi thật do các Task đảm nhiệm. DAG được khai báo hoàn toàn bằng Python — file .py này chính là "cấu hình như code" (configuration as code), cho phép bạn dùng vòng lặp, hàm, biến để sinh pipeline động.

Một DAG được xác định bởi ba nhóm thông tin: định danh (dag_id), lịch chạy (schedule, start_date...), và nội dung (các task cùng quan hệ phụ thuộc).

Task và Task Instance

Task là một node trong DAG — một đơn vị công việc: chạy một câu SQL, gọi một API, chạy một hàm Python, khởi động một job Spark. Task được tạo ra từ Operator (xem bài Operators & Hooks) hoặc từ hàm được trang trí bằng @task.

Cần phân biệt rạch ròi hai tầng khái niệm:

• Task: định nghĩa tĩnh trong code — "hãy làm việc X".
• Task Instance (TI): một lần chạy cụ thể của task đó, gắn với một khoảng thời gian dữ liệu cụ thể. Nếu DAG chạy mỗi ngày trong 30 ngày, một task sẽ có 30 task instance.

Task instance đi qua nhiều trạng thái trong vòng đời:

Trạng thái upstream_failed rất hay gặp: khi một task cha lỗi, mọi task con mặc định sẽ bị đánh dấu upstream_failed và không chạy — đó là hành vi bảo vệ để tránh xử lý trên dữ liệu chưa hoàn chỉnh.

Hai cách viết DAG

Airflow 3 hỗ trợ song song hai phong cách. Bạn có thể trộn cả hai trong cùng một DAG, nhưng nên chọn một phong cách chủ đạo cho dễ đọc.

Cách cổ điển (classic) — Operator + set dependency

Đây là phong cách lâu đời: bạn khởi tạo trực tiếp các đối tượng Operator, rồi nối chúng bằng toán tử >>.

# (minh hoạ) DAG cổ điển: khai báo Operator rồi set phụ thuộc
from airflow import DAG
from airflow.providers.standard.operators.bash import BashOperator
from airflow.providers.standard.operators.empty import EmptyOperator
import pendulum

with DAG(
    dag_id="etl_classic",
    schedule="0 6 * * *",              # 6h sáng mỗi ngày
    start_date=pendulum.datetime(2026, 1, 1, tz="Asia/Ho_Chi_Minh"),
    catchup=False,
    tags=["etl", "daily"],
) as dag:

    start = EmptyOperator(task_id="start")

    extract = BashOperator(
        task_id="extract",
        bash_command="python /opt/jobs/extract.py --date {{ ds }}",
    )

    transform = BashOperator(
        task_id="transform",
        bash_command="python /opt/jobs/transform.py --date {{ ds }}",
    )

    load = BashOperator(
        task_id="load",
        bash_command="python /opt/jobs/load.py --date {{ ds }}",
    )

    start >> extract >> transform >> load

{{ ds }} là một biến template của Airflow (Jinja), sẽ được thay bằng ngày dữ liệu (dạng YYYY-MM-DD) tại thời điểm chạy. Đây là cách truyền "ngày cần xử lý" vào job mà không hardcode.

TaskFlow API — @dag và @task decorator

Phong cách hiện đại, ra đời từ Airflow 2.0 và được khuyến nghị cho code Python thuần. Bạn viết hàm Python bình thường, gắn decorator, và dữ liệu tự chảy qua XCom giữa các task chỉ bằng cách truyền tham số hàm.

# (minh hoạ) DAG dùng TaskFlow API
from airflow.sdk import dag, task
import pendulum

@dag(
    schedule="0 6 * * *",
    start_date=pendulum.datetime(2026, 1, 1, tz="Asia/Ho_Chi_Minh"),
    catchup=False,
    tags=["etl", "daily"],
)
def etl_taskflow():

    @task
    def extract() -> dict:
        # trả về giá trị -> tự động đẩy vào XCom
        return {"rows": [1, 2, 3], "source": "core_banking"}

    @task
    def transform(data: dict) -> int:
        return sum(data["rows"])

    @task
    def load(total: int) -> None:
        print(f"Nạp kết quả tổng = {total}")

    # gọi hàm như Python thường; Airflow tự dựng phụ thuộc
    raw = extract()
    total = transform(raw)
    load(total)

etl_taskflow()

Chú ý sự khác biệt cốt lõi: ở TaskFlow bạn không cần viết >> giữa extract, transform, load. Airflow suy ra phụ thuộc từ việc transform(raw) nhận đầu ra của extract(). Việc truyền dữ liệu giữa các task được lo bởi XCom một cách trong suốt — chi tiết ở bài TaskFlow & XCom.

Trong Airflow 3, các thành phần để viết DAG (như dag, task) được import từ gói airflow.sdk — đây là Task SDK mới tách biệt khỏi lõi scheduler.

Thiết lập phụ thuộc giữa các Task

Phụ thuộc quyết định thứ tự và cấu trúc đồ thị. Có vài cách biểu diễn:

Toán tử dịch chuyển bit >> và << (phổ biến nhất):

# (minh hoạ)
a >> b >> c            # a rồi b rồi c
c << b << a            # tương đương, viết ngược
a >> [b, c]            # fan-out: a xong thì cả b và c chạy song song
[b, c] >> d            # fan-in: d chờ cả b và c xong

chain() — nối tuần tự nhiều task, kể cả nối song song theo cặp:

# (minh hoạ)
from airflow.sdk import chain

chain(a, b, c, d)              # a >> b >> c >> d
chain(a, [b, c], [d, e], f)    # a >> b,c ; b>>d, c>>e ; d,e >> f

cross_downstream() — nối chéo mọi phần tử của nhóm trước tới mọi phần tử nhóm sau:

# (minh hoạ)
from airflow.sdk import cross_downstream

cross_downstream([a, b], [c, d])   # a>>c, a>>d, b>>c, b>>d

Hai mô hình cấu trúc bạn sẽ gặp liên tục:

• Fan-out: một task kích hoạt nhiều task song song (ví dụ extract xong thì nạp song song vào 3 bảng).
• Fan-in: nhiều task hội tụ vào một task (ví dụ chờ cả 3 nguồn về xong mới chạy reconciliation).

Sơ đồ trên minh hoạ một DAG có fan-out tại extract_core (tỏa ra 3 nhánh transform) và fan-in tại load_warehouse (hội tụ 3 nhánh), sau đó kiểm tra chất lượng dữ liệu trước khi kết thúc.

Lịch chạy: schedule, start_date, catchup

schedule

Tham số schedule quyết định DAG chạy khi nào. Airflow 3 chấp nhận nhiều dạng:

• Cron string: "0 6 * * *" (6h sáng mỗi ngày).
• Preset: "@daily", "@hourly", "@weekly", "@monthly", "@once".
• timedelta: timedelta(hours=4) — chạy mỗi 4 giờ tính từ lần chạy trước.
• None: DAG không tự chạy theo lịch, chỉ trigger thủ công hoặc từ DAG khác.
• Asset (data-aware): DAG chạy khi một asset (dữ liệu) được cập nhật — cơ chế thay thế và mở rộng cho Dataset của Airflow 2. Chi tiết ở bài Scheduling & Assets.

start_date

Mốc thời gian DAG bắt đầu có hiệu lực. Nên dùng datetime có timezone (qua pendulum) và cố định — đừng bao giờ dùng datetime.now() làm start_date, vì nó thay đổi mỗi lần parse file khiến scheduler không bao giờ tìm được mốc bắt đầu.

catchup — thay đổi lớn ở Airflow 3

catchup quyết định điều gì xảy ra khi start_date ở trong quá khứ. Nếu catchup=True, Airflow sẽ chạy bù tất cả các khoảng thời gian đã lỡ từ start_date đến hiện tại — có thể tạo ra hàng trăm dag run cùng lúc.

Ở Airflow 3, mặc định của catchup đã đổi thành False (Airflow 2 mặc định là True). Đây là thay đổi an toàn hơn: bạn không còn bị "lũ" dag run bất ngờ nếu quên set. Nếu thật sự cần hành vi cũ, có thể chỉnh config catchup_by_default = True. Khuyến nghị chung: luôn để catchup=False và chủ động backfill khi cần.

Backfill — chạy bù có kiểm soát

Backfill là việc chủ động chạy DAG cho một khoảng thời gian trong quá khứ — ví dụ vừa sửa logic transform và cần tính lại 7 ngày dữ liệu cũ. Khác với catchup (tự động khi khởi tạo DAG), backfill là hành động có chủ đích.

Ở Airflow 3, backfill được quản lý bởi scheduler và có thể khởi tạo qua CLI, UI hoặc API — cải tiến đáng kể so với 2.x nơi backfill chỉ chạy qua CLI như một tiến trình riêng. Nhờ vậy backfill hiển thị và điều khiển được ngay trên giao diện, ổn định hơn trong môi trường thực.

# (minh hoạ) backfill qua CLI
airflow backfill create \
  --dag-id etl_taskflow \
  --from-date 2026-06-01 \
  --to-date 2026-06-07

Data interval và cái chết của execution_date

Đây là khái niệm khiến nhiều người mới bối rối, nhưng cực kỳ quan trọng.

Airflow không nghĩ theo "thời điểm chạy" mà theo khoảng thời gian dữ liệu (data interval) mà một lần chạy chịu trách nhiệm xử lý. Mỗi dag run gắn với một khoảng [data_interval_start, data_interval_end).

Ví dụ trực giác: một DAG @daily chạy cho dữ liệu của ngày 30/06 thường sẽ thực thi vào sáng 01/07 — vì phải chờ hết ngày 30/06 mới có đủ dữ liệu. Khoảng dữ liệu là "cả ngày 30/06", còn thời điểm chạy vật lý là 01/07.

Các biến bạn dùng trong template và context:

• logical_date: nhãn thời gian định danh cho dag run (trước đây gọi là execution_date).
• data_interval_start và data_interval_end: hai đầu của khoảng dữ liệu.
• ds: logical_date dạng YYYY-MM-DD — biến hay dùng nhất để tham số hóa job.

Airflow 3 đã bỏ hẳn execution_date. Cái tên execution_date gây hiểu nhầm suốt nhiều năm vì nó nghe như "thời điểm chạy" nhưng thực chất là mốc logic của khoảng dữ liệu. Từ Airflow 3, bạn dùng logical_date (và các biến data_interval_*) thay thế.

Một thay đổi kỹ thuật đi kèm: timetable mặc định cho cron đổi từ CronDataIntervalTimetable sang CronTriggerTimetable. Hệ quả là với lịch cron, data_interval_start và data_interval_end mặc định bằng chính logical_date, và logical_date giờ là thời điểm sau đó task thực sự chạy. Ngoài ra Airflow 3 cho phép dag run có logical_date = None, giúp cùng một DAG khởi chạy nhiều lần tại cùng thời điểm (hữu ích cho các use case như inference AI).

Bài học thực hành: luôn tham số hóa job theo ds / data_interval_*, không dùng "hôm nay" của hệ thống. Nhờ vậy chạy lại (rerun/backfill) một ngày cũ sẽ cho ra đúng kết quả của ngày đó.

Retry và xử lý lỗi

Task lỗi là chuyện bình thường trong hệ phân tán — mạng chập chờn, nguồn quá tải. Airflow cung cấp cơ chế retry ở mức task:

• retries: số lần thử lại (ví dụ 3).
• retry_delay: khoảng chờ giữa các lần thử (ví dụ timedelta(minutes=5)).
• retry_exponential_backoff: nếu True, khoảng chờ tăng theo cấp số nhân (5 phút, 10 phút, 20 phút...) — tránh dồn dập tấn công một nguồn đang gặp sự cố.
• max_retry_delay: trần cho backoff.
• execution_timeout: giới hạn thời gian chạy; quá hạn thì task bị kill và tính là fail.

# (minh hoạ) cấu hình retry qua default_args
from datetime import timedelta

default_args = {
    "owner": "data-eng",
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
    "retry_exponential_backoff": True,
    "max_retry_delay": timedelta(minutes=30),
}

Trong vòng lượt retry, task ở trạng thái up_for_retry; hết lượt mà vẫn lỗi thì chuyển sang failed và kích hoạt các cơ chế cảnh báo (email, callback, on_failure_callback).

Idempotency & atomicity

Hai tính chất này là nền tảng để pipeline "chạy lại được mà không hỏng dữ liệu" — chủ đề sẽ đào sâu ở bài best practices.

• Idempotency (bất biến khi lặp lại): chạy cùng một task với cùng data interval nhiều lần phải cho cùng một kết quả cuối, không tạo bản ghi trùng. Cách phổ biến: task load nên xóa/ghi đè đúng partition ngày trước khi ghi (delete-then-insert theo ds), thay vì "append" mù. Nhờ đó backfill và retry đều an toàn.
• Atomicity (nguyên tử): một task hoặc hoàn thành trọn vẹn, hoặc không để lại tác dụng phụ nào. Tránh cảnh task ghi được nửa dữ liệu rồi lỗi, để lại trạng thái dở dang. Kỹ thuật: ghi vào bảng/thư mục tạm, rồi mới "swap" nguyên tử sang đích khi đã chắc chắn thành công.

Hai tính chất này chính là lý do vì sao ta luôn tham số hóa theo data interval: chỉ khi mỗi lần chạy "sở hữu" đúng một khoảng dữ liệu xác định thì mới làm được idempotent một cách sạch sẽ.

Cấu trúc thư mục và tổ chức DAG

Airflow tự động quét thư mục cấu hình bởi dags_folder (mặc định $AIRFLOW_HOME/dags). Mọi file .py trong đó được parse để tìm object DAG. Một cấu trúc production thường trông như sau:

dags/
├── etl_core_banking.py        # 1 file = 1 (hoặc vài) DAG
├── etl_customer_360.py
├── common/                    # code dùng chung, KHÔNG chứa DAG
│   ├── sql/
│   └── helpers.py
└── .airflowignore             # loại trừ file/thư mục khỏi việc parse

Vài nguyên tắc tổ chức:

• Mỗi file parse phải nhanh — Airflow parse toàn bộ dags_folder liên tục. Đừng gọi API hay query DB ở top-level của file DAG (chỉ chạy khi thực thi task, không phải khi parse).
• default_args: dict các tham số áp dụng cho mọi task trong DAG (owner, retries, retry_delay...). Giúp không lặp lại cấu hình ở từng task.
• tags: gắn nhãn để lọc/nhóm DAG trên UI (["etl", "daily", "core-banking"]).
• doc_md: gắn tài liệu Markdown ngay vào DAG hoặc task; nội dung hiển thị trên UI — cực kỳ hữu ích cho người trực vận hành.

# (minh hoạ) DAG production đầy đủ với default_args, tags, doc_md
from airflow.sdk import dag, task
from datetime import timedelta
import pendulum

default_args = {
    "owner": "data-eng",
    "retries": 2,
    "retry_delay": timedelta(minutes=5),
}

@dag(
    dag_id="etl_core_banking_daily",
    schedule="0 6 * * *",
    start_date=pendulum.datetime(2026, 1, 1, tz="Asia/Ho_Chi_Minh"),
    catchup=False,
    default_args=default_args,
    tags=["etl", "daily", "core-banking"],
    doc_md="""
### ETL Core Banking (EOD)
Pipeline chạy cuối ngày: extract giao dịch từ core, transform, nạp vào warehouse.
Owner: data-eng. Liên hệ #data-oncall nếu lỗi.
""",
)
def etl_core_banking_daily():

    @task
    def extract(data_interval_start=None):
        # data_interval_start được inject tự động từ context
        return f"trích dữ liệu cho {data_interval_start}"

    @task
    def transform(payload: str) -> str:
        return payload.upper()

    @task
    def load(payload: str) -> None:
        print(payload)

    load(transform(extract()))

etl_core_banking_daily()

Use case thực tế: pipeline ETL cuối ngày

Hãy hình dung một pipeline EOD (end-of-day) cho ngân hàng, chạy 6h sáng mỗi ngày để xử lý dữ liệu của ngày hôm trước:

1. extract_core — trích toàn bộ giao dịch, khách hàng, tài khoản của ngày ds từ core banking sang vùng staging.
2. fan-out 3 nhánh transform — song song chuẩn hóa transform_txn, transform_customer, transform_account. Vì độc lập nhau nên chạy song song để rút ngắn thời gian.
3. fan-in load_warehouse — chờ cả 3 nhánh xong, nạp vào warehouse theo partition ngày ds (delete-then-insert để idempotent).
4. data_quality_check — kiểm tra số bản ghi, tính khớp số dư (reconciliation). Nếu fail thì dừng và cảnh báo.
5. end — chốt pipeline; có thể phát một asset để kích hoạt các DAG hạ nguồn (báo cáo, dbt).

Đặc điểm production của pipeline này:

• catchup=False, và nếu cần tính lại lịch sử thì dùng backfill có kiểm soát.
• Mọi job tham số hóa theo ds — chạy lại ngày 25/06 luôn cho đúng dữ liệu ngày 25/06.
• retries=2 cho các task gọi mạng; data_quality_check không nên tự retry mù mà nên fail rõ ràng để người kiểm tra.
• Task load idempotent theo partition, task ghi qua bảng tạm rồi swap để đảm bảo atomicity.

Đây chính là khung xương của phần lớn pipeline batch bạn sẽ gặp trong thực tế — phần còn lại chỉ là thay đổi Operator và logic nghiệp vụ bên trong.

Ghi nhớ

• DAG = đồ thị có hướng không chu trình, mô tả cấu trúc và lịch; Task làm việc thật; Task Instance là một lần chạy cụ thể gắn với một data interval.
• Hai cách viết: TaskFlow (@dag/@task, dữ liệu tự chảy qua XCom, khuyến nghị cho Python thuần) và cổ điển (Operator + >>). Trong Airflow 3, import từ airflow.sdk.
• Phụ thuộc: >>/<<, chain(), cross_downstream(); nắm vững fan-out và fan-in.
• Lịch: schedule (cron/preset/timedelta/None/Asset), start_date cố định có timezone. Airflow 3 mặc định catchup=False — nên giữ nguyên và dùng backfill (qua UI/API/CLI, do scheduler quản lý) khi cần chạy bù.
• Data interval là trái tim của tư duy Airflow: tham số hóa theo logical_date/ds/data_interval_*. Airflow 3 bỏ execution_date, dùng logical_date.
• Xử lý lỗi bằng retries, retry_delay, retry_exponential_backoff; hiểu vòng đời trạng thái (đặc biệt up_for_retry, upstream_failed).
• Idempotency và atomicity là điều kiện để rerun/backfill an toàn — thiết kế task ngay từ đầu theo hai tính chất này.

Bài tiếp theo đi sâu vào Operators & Hooks — công cụ để Task thực sự "làm việc" với các hệ thống bên ngoài; sau đó là TaskFlow & XCom và Scheduling & Assets.