DAG là gì

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

• DAG là gì và vì sao phải là đồ thị không chu trình.
• Các thành phần của Airflow: scheduler, executor, metadata database.
• Đọc đúng data_interval thay vì nhầm với giờ chạy.
• Viết DAG đầu tiên bằng TaskFlow API.
• Những hiểu nhầm khiến pipeline chạy sai ngày.

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

DAG (Directed Acyclic Graph) là tập các task có hướng và không có chu trình.

extract_orders → transform_orders → load_orders

• Có hướng: extract chạy trước transform.
• Không chu trình: không có đường quay lại extract. Nếu có, Airflow không biết bắt đầu từ đâu.

Mỗi lần DAG chạy gọi là một DAG run, gắn với một logical date (khoảng dữ liệu), không phải thời điểm bấm chạy.

Các thành phần

Nếu metadata DB mất, Airflow mất toàn bộ lịch sử chạy. Sao lưu DB quan trọng hơn sao lưu file DAG.

Data interval

Đây là khái niệm gây nhầm lẫn nhiều nhất.

DAG chạy lúc 02:00 ngày 2/3 với lịch hằng ngày xử lý dữ liệu của ngày 1/3:

data_interval_start = 2026-03-01 00:00
data_interval_end   = 2026-03-02 00:00
logical date        = 2026-03-01 (đầu khoảng)

Quy tắc: một lần chạy chỉ được xử lý dữ liệu đã hoàn tất tại thời điểm nó bắt đầu. Vì vậy nó xử lý khoảng phía trước, không phải khoảng đang diễn ra.

Nhầm lẫn phổ biến: dùng datetime.now() trong task. Mỗi lần chạy lại cho kết quả khác và không tái lập được. Luôn dùng data_interval_start hoặc ds.

DAG đầu tiên

from datetime import datetime, timedelta

from airflow.decorators import dag, task


@dag(
    dag_id="daily_orders",
    schedule="0 2 * * *",
    start_date=datetime(2026, 1, 1),
    catchup=False,
    default_args={
        "retries": 2,
        "retry_delay": timedelta(minutes=5),
    },
    tags=["orders"],
)
def daily_orders():
    @task
    def extract(**context):
        ngay = context["data_interval_start"].date()
        return {"ngay": str(ngay), "so_dong": 12000}

    @task
    def transform(payload):
        return {**payload, "so_dong": payload["so_dong"]}

    @task
    def load(payload):
        print(f"Nạp {payload['so_dong']} dòng cho ngày {payload['ngay']}")

    load(transform(extract()))


daily_orders()

Điểm cần chú ý:

• catchup=False để không tự chạy bù toàn bộ quá khứ từ start_date.
• retries đặt ở default_args áp cho mọi task.
• TaskFlow suy ra phụ thuộc từ cách gọi hàm: load(transform(extract())) tạo chuỗi ba task.
• data_interval_start lấy từ context, không dùng now().

DAG không chu trình

Ví dụ sai:

extract → transform → load → transform

Airflow sẽ báo lỗi chu trình khi parse. Cách sửa: tách bước kiểm tra thành task riêng, hoặc gộp logic.

Kiểm tra DAG trước khi đưa lên

airflow dags list
airflow tasks list daily_orders
airflow dags test daily_orders 2026-03-01

dags test chạy toàn bộ DAG cho một logical date mà không ghi vào lịch sử production. Đây là cách kiểm tra nhanh nhất trước khi bật lịch.

Lỗi thường gặp

• Dùng datetime.now(): kết quả không tái lập.
• Nhầm logical date với giờ chạy: xử lý sai ngày.
• Để catchup=True mặc định: chạy bù hàng trăm ngày khi mới bật DAG.
• Đặt logic nặng ở cấp DAG: file parse chậm, scheduler quá tải.
• Một DAG làm quá nhiều việc: khó theo dõi và chạy lại.
• Không đặt start_date: DAG không được lập lịch.

Thực hành

1. Vẽ DAG ba task cho pipeline đơn hàng.
2. Xác định data_interval_start cho lần chạy 2/3 lúc 02:00.
3. Viết DAG TaskFlow với retries=2.
4. Chạy airflow dags test cho một ngày cụ thể.
5. Giải thích vì sao catchup=True có thể gây nguy hiểm.

Lỗi thường gặp

Tóm tắt

• DAG là đồ thị có hướng, không chu trình.
• Logical date là khoảng dữ liệu, không phải giờ chạy.
• Dùng data_interval_start, không dùng datetime.now().
• catchup=False khi mới bật DAG.
• Kiểm tra bằng airflow dags test trước khi bật lịch.