Airflow 6 — Lập lịch, Timetables & Assets (event-driven)

Lập lịch (scheduling) là trái tim của Airflow. Nhưng nếu chỉ dừng ở việc gõ một chuỗi cron rồi để scheduler "nổ" DAG mỗi ngày, bạn mới dùng khoảng 20% sức mạnh thật. Airflow 3.x mở ra hai hướng lớn: Timetables cho những lịch không thể diễn tả bằng cron thuần, và Assets (tên mới của Datasets) cho lập lịch theo dữ liệu thay vì theo giờ — kết hợp với event-driven scheduling để react gần real-time với message từ hệ thống bên ngoài.

Bài này đi sâu vào cơ chế bên trong scheduler, cách nó sinh ra DAG run, rồi tới Timetables và Assets. Trước khi đọc, nên nắm chắc DAG & Task và Operators & Hooks.

1. Scheduler tạo DAG run như thế nào

Mỗi DAG có hai thứ quyết định lịch chạy: tham số schedule và start_date. Scheduler liên tục quét (parse) các DAG, và với mỗi DAG nó tính: "Đã tới lúc tạo một DAG run mới chưa?". Câu trả lời dựa trên khái niệm cốt lõi — data interval (khoảng dữ liệu).

Data interval và logical_date

Airflow không nghĩ theo kiểu "chạy lúc 2 giờ sáng", mà theo kiểu "xử lý khoảng dữ liệu nào". Mỗi DAG run gắn với một data interval — một khoảng thời gian đóng-mở [data_interval_start, data_interval_end) mà run đó chịu trách nhiệm xử lý.

Ví dụ một DAG @daily: run cho ngày 2026-06-30 có:

• data_interval_start = 2026-06-30 00:00
• data_interval_end = 2026-07-01 00:00
• Run này chỉ thực sự khởi động sau khi khoảng đó kết thúc, tức khoảng sau nửa đêm 01/07.

Đây là điểm gây bối rối kinh điển: DAG "của ngày 30" chạy vào ngày 01. Lý do rất thực tế — muốn xử lý dữ liệu của cả một ngày thì phải đợi ngày đó trôi qua đã.

Cùng với đó là logical_date (trước đây gọi là execution_date). Đây là một mốc thời gian định danh logic của run, KHÔNG phải "thời điểm chạy vật lý". Trong Airflow 3, logical_date được hiểu là thời điểm mà sau đó run mới thật sự chạy, và với timetable cron mặc định mới (CronTriggerTimetable) thì data_interval_start, data_interval_end và logical_date mặc định trùng nhau — khác với hành vi Airflow 2 cũ.

Airflow 3 bỏ hẳn execution_date. Trong code và trong context template, hãy dùng logical_date, data_interval_start, data_interval_end. Các key cũ như execution_date, next_ds, prev_ds... đã bị gỡ.

Catchup và Backfill

Hai khái niệm hay bị nhầm:

• Catchup: nếu start_date nằm trong quá khứ và catchup=True, scheduler sẽ tự động sinh ra tất cả các run bị bỏ lỡ từ start_date tới hiện tại. Với DAG chạy nhiều lần/ngày trong nhiều tháng, đây là quả bom — thường ta để catchup=False (mặc định khuyến nghị) để chỉ chạy run gần nhất.
• Backfill: chạy lại một cách chủ động cho một khoảng quá khứ cụ thể (ví dụ code mới sửa xong, cần tính lại 3 tháng dữ liệu). Trong Airflow 3, backfill được nâng cấp thành tính năng first-class, quản lý qua API/UI/CLI chứ không chỉ là lệnh CLI rời rạc như trước.

2. Các kiểu schedule

Tham số schedule chấp nhận nhiều dạng:

Khác biệt tinh tế giữa cron và timedelta: cron neo vào mốc thời gian tuyệt đối (đúng 2h sáng), còn timedelta neo vào khoảng cách so với run trước (cứ 6 tiếng lại chạy, không quan tâm đồng hồ chỉ mấy giờ).

# (minh hoạ) — DAG chạy 2h sáng mỗi ngày, không catchup
from airflow.sdk import DAG
from airflow.providers.standard.operators.python import PythonOperator
import pendulum

with DAG(
    dag_id="daily_etl",
    schedule="0 2 * * *",
    start_date=pendulum.datetime(2026, 1, 1, tz="Asia/Ho_Chi_Minh"),
    catchup=False,
    tags=["etl"],
):
    PythonOperator(task_id="run", python_callable=lambda: None)

Lưu ý start_date nên gắn timezone rõ ràng — Airflow xử lý DST và lệch múi giờ dựa vào đó.

3. Timetables — khi cron không đủ

Cron rất mạnh nhưng không diễn tả được các quy tắc như: "chỉ chạy ngày làm việc, bỏ cuối tuần và ngày lễ", "chạy vào ngày làm việc cuối cùng của tháng", hay "data interval kéo dài từ thứ Sáu tới thứ Hai". Đây là lúc dùng Timetable.

Về bản chất, mọi schedule (kể cả cron) đều được Airflow chuyển thành một Timetable ở bên trong. Một Timetable có hai nhiệm vụ:

1. next_dagrun_info(...) — cho scheduler biết run tiếp theo rơi vào khoảng dữ liệu nào (trả về DagRunInfo chứa data_interval và mốc chạy).
2. infer_manual_data_interval(...) — khi ai đó trigger tay, tính data interval hợp lý cho run đó.

Built-in timetables

Airflow cung cấp sẵn nhiều timetable, không phải cái gì cũng phải tự viết:

• CronTriggerTimetable — timetable mặc định diễn giải chuỗi cron trong Airflow 3 (thay cho CronDataIntervalTimetable cũ). Đây là lý do logical_date giờ trùng data interval.
• CronDataIntervalTimetable — bản cũ vẫn còn, tính data interval theo kiểu "chạy sau khi khoảng kết thúc".
• DeltaDataIntervalTimetable — cho timedelta.
• EventsTimetable — chạy theo một danh sách các mốc sự kiện định trước (ví dụ ngày mở cửa thị trường, lịch phát hành báo cáo), không theo chu kỳ đều.
• AssetTriggeredTimetable — được dùng ngầm khi bạn lập lịch theo Asset.

Viết một Timetable tuỳ biến

Khi built-in không đủ, kế thừa lớp Timetable và cài đặt hai phương thức trên. Dưới đây là bản rút gọn: chỉ chạy các ngày làm việc (Thứ Hai–Thứ Sáu), data interval là một ngày.

# (minh hoạ) — Timetable rút gọn: chỉ chạy ngày làm việc
from airflow.timetables.base import Timetable, DagRunInfo, DataInterval, TimeRestriction
from pendulum import DateTime, Duration
from typing import Optional

class WorkdayTimetable(Timetable):
    def infer_manual_data_interval(self, run_after: DateTime) -> DataInterval:
        start = run_after.start_of("day")
        return DataInterval(start=start, end=start + Duration(days=1))

    def next_dagrun_info(
        self,
        *,
        last_automated_data_interval: Optional[DataInterval],
        restriction: TimeRestriction,
    ) -> Optional[DagRunInfo]:
        # điểm neo: sau interval trước, hoặc từ start_date
        if last_automated_data_interval is not None:
            next_start = last_automated_data_interval.end
        else:
            if restriction.earliest is None:
                return None
            next_start = restriction.earliest.start_of("day")

        # nhảy qua Thứ Bảy (5) và Chủ Nhật (6)
        while next_start.weekday() >= 5:
            next_start = next_start + Duration(days=1)

        if restriction.latest is not None and next_start > restriction.latest:
            return None

        end = next_start + Duration(days=1)
        return DagRunInfo.interval(start=next_start, end=end)

Dùng nó: with DAG(dag_id="workday_dag", schedule=WorkdayTimetable(), ...). Để dùng lâu dài trên toàn hệ thống, đóng gói timetable trong một plugin để scheduler nhận diện được khi serialize DAG.

TimeRestriction mang các ràng buộc earliest (thường là start_date), latest (end_date) và catchup. Timetable phải tôn trọng chúng — trả None nghĩa là "không còn run nào để lên lịch".

4. Assets — lập lịch theo dữ liệu (trước gọi Datasets)

Đây là thay đổi tư duy quan trọng nhất. Thay vì hỏi "khi nào tới giờ?", ta hỏi "khi nào dữ liệu sẵn sàng?".

Trong Airflow 2.4 khái niệm này ra đời với tên Datasets. Airflow 3 đổi tên thành Assets (AIP-75), mở rộng thêm cú pháp asset-centric, watchers và event-driven.

Khái niệm

Một Asset đại diện cho một đối tượng dữ liệu logic — thường là một file S3, một bảng, một partition — định danh bằng một URI:

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

sales_asset = Asset("s3://datalake/sales/daily.parquet")
# hoặc đặt tên riêng cho dễ đọc
sales_asset = Asset(uri="s3://datalake/sales/daily.parquet", name="daily_sales")

Producer → Consumer

• Producer: DAG nào tạo/cập nhật dữ liệu sẽ khai báo asset đó trong outlets của task. Khi task chạy xong thành công, Airflow ghi nhận một AssetEvent (asset vừa được cập nhật).
• Consumer: DAG nào phụ thuộc dữ liệu đó chỉ cần đặt schedule=[asset]. Airflow sẽ tự chạy consumer ngay sau khi producer cập nhật asset — không cần cron, không cần sensor chờ.

# (minh hoạ) — Producer cập nhật asset
from airflow.sdk import DAG, Asset
from airflow.providers.standard.operators.python import PythonOperator

sales_asset = Asset("s3://datalake/sales/daily.parquet")

with DAG(dag_id="produce_sales", schedule="0 3 * * *", catchup=False):
    PythonOperator(
        task_id="write_sales",
        outlets=[sales_asset],          # <-- khai báo output là asset
        python_callable=lambda: None,
    )

# (minh hoạ) — Consumer chạy theo asset, không cần cron
from airflow.sdk import DAG
from airflow.providers.standard.operators.python import PythonOperator

with DAG(dag_id="build_report", schedule=[sales_asset], catchup=False):
    PythonOperator(task_id="report", python_callable=lambda: None)

Airflow 3 còn có cú pháp decorator gọn hơn:

# (minh hoạ) — định nghĩa asset bằng decorator
from airflow.sdk import asset

@asset(uri="s3://datalake/sales/daily.parquet", schedule="@daily")
def daily_sales():
    ...  # thân hàm ghi dữ liệu ra asset

Kết hợp nhiều asset (AND / OR)

Consumer có thể phụ thuộc nhiều asset:

• AND (chạy khi tất cả asset đã cập nhật ít nhất một lần kể từ run trước): dùng list hoặc toán tử &.

  # (minh hoạ)
  with DAG(dag_id="join_dag", schedule=[asset_a, asset_b]):  # AND
      ...
  # tương đương: schedule=(asset_a & asset_b)
  

• OR (chạy khi bất kỳ asset nào cập nhật): dùng AssetAny hoặc toán tử |.

  # (minh hoạ)
  from airflow.sdk import AssetAny, AssetAll
  with DAG(dag_id="either_dag", schedule=AssetAny(asset_a, asset_b)):  # OR
      ...
  

  AssetAll biểu diễn AND tường minh; có thể lồng nhau để tạo biểu thức logic phức tạp, ví dụ (asset_a & asset_b) | asset_c.

Với AND, cần nhấn mạnh ngữ nghĩa: DAG chỉ chạy sau khi mọi asset đầu vào đã được cập nhật ít nhất một lần kể từ lần chạy trước của DAG — tránh chạy dở dang khi mới một nửa upstream xong.

Sơ đồ luồng

5. Event-driven scheduling (Airflow 3)

Assets cơ bản react với sự kiện nội bộ Airflow (một task trong Airflow khác vừa cập nhật asset). Event-driven scheduling đi xa hơn: react với sự kiện từ hệ thống bên ngoài — một message rơi vào hàng đợi SQS, một record vào Kafka, một thông báo Pub/Sub — để kích hoạt DAG gần real-time.

Cơ chế xoay quanh ba mảnh ghép:

1. Trigger — một trigger kế thừa BaseEventTrigger lắng nghe nguồn ngoài. Provider apache-airflow-providers-common-messaging cung cấp MessageQueueTrigger với "common message bus" trừu tượng hoá nhiều backend (SQS, Kafka, Pub/Sub...).
2. AssetWatcher — bọc một trigger và gắn nó vào một asset. Khi trigger phát ra TriggerEvent, watcher cập nhật asset (sinh AssetEvent).
3. Consumer DAG — vẫn chỉ schedule=[asset] như thường. Nó không biết message đến từ đâu; nó chỉ thấy "asset được cập nhật".

# (minh hoạ) — asset lắng nghe message SQS
from airflow.sdk import Asset, AssetWatcher
from airflow.providers.common.messaging.triggers.msg_queue import MessageQueueTrigger

trigger = MessageQueueTrigger(queue="https://sqs.ap-southeast-1.amazonaws.com/123/my-queue")

incoming = Asset(
    name="incoming_files",
    watchers=[AssetWatcher(name="sqs_watcher", trigger=trigger)],
)

# DAG này chạy mỗi khi có message tới hàng đợi
# with DAG(dag_id="react_to_msg", schedule=[incoming]): ...

Trigger chạy trong triggerer (thành phần async của Airflow), nên hàng ngàn watcher có thể "ngồi chờ" mà tốn rất ít tài nguyên — không đốt worker slot như sensor kiểu poke.

Ranh giới với streaming thuần

Đây là điểm phải rất tỉnh táo. Event-driven scheduling không biến Airflow thành Flink/Kafka Streams:

• Mỗi sự kiện thường tạo ra một DAG run — phù hợp với micro-batch, tần suất giây/phút, không phải hàng chục nghìn message/giây.
• Airflow điều phối workflow; nếu bạn cần xử lý dòng liên tục với latency mili-giây, hãy dùng công cụ streaming và để Airflow điều phối các job đó (xem 12-cdc-realtime và 05-kafka-streaming).
• Cảnh báo tài liệu: tránh trigger dựa trên trạng thái tồn tại lâu dài (file đã tồn tại, job đã xong) vì điều kiện luôn đúng → rescheduling vô hạn. Event-driven hợp với sự kiện tức thời (message tới), không hợp với trạng thái bền.

6. SLA cũ và Deadline Alerts mới

Cơ chế SLA của Airflow 2 (sla=... trên task) đã bị gỡ ở Airflow 3.0, và từ 3.1 được thay bằng Deadline Alerts (AIP-86). Khác biệt về khái niệm:

• SLA cũ chỉ đánh giá sau khi task/DAG kết thúc, thường báo trễ và khó cấu hình.
• Deadline Alerts đặt một mốc thời hạn tương đối so với một điểm tham chiếu (ví dụ DeadlineReference.DAGRUN_LOGICAL_DATE + khoảng thời gian). Scheduler kiểm tra định kỳ và gọi callback ngay khi quá hạn, không cần đợi DAG chạy xong — nên bạn được cảnh báo trong lúc sự cố đang diễn ra chứ không phải sau đó.

Ở mức thực hành: khi migrate lên Airflow 3, gỡ mọi sla= và định nghĩa lại kỳ vọng bằng Deadline Alerts. Đây là thay đổi hành vi, không chỉ đổi tên.

7. Use case thực tế: bỏ chuỗi sensor, dùng Asset

Bài toán kinh điển: pipeline B cần dữ liệu do pipeline A tạo ra. Cách làm cũ (Airflow 2):

• Cho B chạy cron lúc 4h sáng, hy vọng A đã xong lúc 3h.
• Hoặc đặt một ExternalTaskSensor/S3KeySensor trong B, poll liên tục chờ A. Sensor này chiếm worker slot, phải tinh chỉnh timeout, và nếu A trễ thì B hoặc chờ mòn mỏi hoặc fail.

Chuỗi này càng dài (A → B → C → D, mỗi bước một sensor chờ bước trước) càng mong manh và tốn tài nguyên.

Cách làm Asset (Airflow 3):

• A khai báo outlets=[sales_asset].
• B đặt schedule=[sales_asset] — bỏ hoàn toàn cron và sensor.
• Khi A xong (dù 3h hay 5h), B tự chạy ngay. C phụ thuộc output của B thì lại schedule=[report_asset], cứ thế nối chuỗi.

Kết quả: pipeline chạy khi dữ liệu sẵn sàng, không sớm không muộn; không worker slot bị sensor giữ; đồ thị phụ thuộc dữ liệu hiện rõ trong UI (trang Assets cho thấy asset nào cập nhật lúc nào, DAG nào tiêu thụ). Khi cần react với nguồn ngoài, gắn AssetWatcher vào asset là pipeline thành gần real-time mà không đổi consumer.

Về cách tổ chức code và trade-off, xem thêm Best practices.

Ghi nhớ

• Data interval là gốc: DAG run gắn với khoảng [start, end); @daily của ngày 30 chạy sau khi ngày 30 kết thúc. Airflow 3 dùng logical_date (mặc định trùng data interval) và đã bỏ execution_date.
• catchup=False gần như luôn là lựa chọn an toàn để tránh sinh hàng loạt run quá khứ.
• Timetable là lớp bên dưới mọi schedule. Cron không diễn tả được (ngày làm việc, lịch bất thường) → dùng built-in như EventsTimetable hoặc viết custom kế thừa Timetable với next_dagrun_info + infer_manual_data_interval.
• Asset = lập lịch theo dữ liệu: producer khai báo outlets, consumer schedule=[asset]. Kết hợp AND (&, list) / OR (AssetAny, |).
• Event-driven = AssetWatcher + MessageQueueTrigger để react với message ngoài (SQS/Kafka/Pub/Sub) gần real-time, nhưng không thay thế streaming thuần; tránh trigger theo trạng thái bền để không rescheduling vô hạn.
• SLA đã chết, Deadline Alerts lên ngôi — callback gọi ngay khi quá hạn, không đợi DAG xong.
• Chuyển từ sensor-chờ-nhau sang Asset làm pipeline gọn, rẻ và đúng thời điểm hơn.