Apache Airflow: Xây dựng và vận hành Data Pipeline thực tế · Bài 3/6. Một pipeline chạy lúc 00:05 ngày 02/01 có thể đang xử lý đơn của ngày 01/01. Nếu code luôn lấy “hôm nay” theo đồng hồ của worker, việc chạy lại vào tuần sau có thể đọc một ngày hoàn toàn khác. DAG vẫn thành công nhưng bạn không còn biết báo cáo đã xử lý đúng dữ liệu nào.

Bài này giữ nguyên API và PostgreSQL ở bài 2, chỉ thay cách xác định phạm vi dữ liệu. Mỗi run xử lý đúng một khoảng ngày tại Việt Nam; cùng khoảng đó có thể được chạy lại hoặc chạy bù mà không phụ thuộc ngày hiện tại của máy. Chúng ta sẽ phân biệt lịch chạy, logical date, data interval, catchup và backfill bằng những con số kiểm tra được.

Lab dùng Airflow 3.1.7 và một CronDataIntervalTimetable được khai báo tường minh. Điểm này quan trọng vì Airflow 3 thay đổi mặc định cron so với nhiều ví dụ Airflow 2: không nên thấy schedule="@daily" rồi mặc định mọi môi trường đều có cách gán khoảng dữ liệu giống nhau. Ví dụ ở bài dùng chính timetable để xác định ngữ nghĩa, không dựa vào một cấu hình mặc định ẩn.

1. Có ít nhất bốn loại thời gian cần phân biệt

Ngày nghiệp vụ là ngày đơn hàng phát sinh theo quy ước doanh nghiệp. Data interval là khoảng dữ liệu một run dự kiến xử lý. Logical date nhận diện run theo quy tắc timetable; nó không phải đồng hồ ghi nhận lúc Python bắt đầu chạy. Thời điểm thực thi thực tế có thể muộn hơn lịch vì worker bận, hệ thống tạm dừng hoặc bạn đang backfill.

Dấu ngoặc vuông bên trái nghĩa là lấy cả mốc đầu. Dấu ngoặc tròn bên phải nghĩa là không lấy mốc cuối. Một đơn đúng 02/01 00:00 sẽ thuộc ngày 02/01, không xuất hiện đồng thời trong cả hai ngày. Đây là quy ước giúp ghép các khoảng liên tiếp không hở và không chồng.

Giả sử bạn backfill dữ liệu 01/01 vào ngày 10/01. Lúc thực thi là 10/01 nhưng API vẫn phải nhận from=01/01 và to=02/01. Thay datetime.now() bằng “ngày hôm qua” cũng chưa giải quyết được: ngày hôm qua của 10/01 là 09/01, không phải ngày cần chạy bù.

2. Múi giờ cần đi cùng ranh giới ngày

Series quy ước ngày bán hàng theo Asia/Ho_Chi_Minh. Một ngày 01/01 tại Việt Nam bắt đầu ở 2026-01-01T00:00:00+07:00 và kết thúc ở 2026-01-02T00:00:00+07:00. Cùng hai thời điểm đó nếu biểu diễn UTC là 2025-12-31T17:00:00+00:00 và 2026-01-01T17:00:00+00:00. Khác chuỗi hiển thị nhưng cùng thời điểm.

where created_at >= :interval_start
  and created_at < :interval_end

Đây là dạng điều kiện minh họa phạm vi, không phải câu SQL chạy độc lập. Code đầy đủ bên dưới truyền tham số cho psycopg2 bằng %s. Tránh ghép timestamp người dùng vào chuỗi SQL. API nhận các timestamp có offset bằng requests.params để dấu +07:00 được URL encode đúng, thay vì nối URL thủ công làm dấu cộng bị diễn giải như khoảng trắng.

PostgreSQL timestamptz lưu một thời điểm và hiển thị theo timezone của session. Việc một công cụ SQL hiện UTC không có nghĩa dữ liệu bị lệch. Khi nhóm theo ngày Việt Nam, dùng created_at AT TIME ZONE 'Asia/Ho_Chi_Minh' trước khi lấy date, thay vì ép created_at::date trong một session chưa rõ timezone.

3. Vì sao dùng CronDataIntervalTimetable tường minh?

Một lịch có thể mang nghĩa “đến 00:00 thì kích hoạt” hoặc “khi khoảng ngày đã khép lại thì xử lý khoảng vừa hoàn tất”. Hai cách nghe gần giống nhau nhưng có thể tạo data interval khác. Trong Airflow 3.1.7, cấu hình mặc định create_cron_data_intervals đã thay đổi; bài này không dựa vào cách diễn giải mặc định của cron string.

from airflow.timetables.interval import CronDataIntervalTimetable

schedule=CronDataIntervalTimetable(
    '0 0 * * *', timezone='Asia/Ho_Chi_Minh'
)

Với timetable này, start_date=01/01 00:00 và catchup=True trong một hệ thống theo dõi từ trước đó: khoảng đầu là ngày 01/01, đủ điều kiện sau 02/01 00:00. Nếu bạn mới triển khai vào tháng 11 với catchup=False như code thực hành, Airflow không tự chạy toàn bộ các ngày từ tháng 1. Vì thế để nghiệm thu fixture cũ, chúng ta sẽ chủ động chạy ngày mẫu hoặc backfill phạm vi nhỏ.

Nếu doanh nghiệp muốn chạy lúc 02:00 nhưng báo cáo ngày lịch hôm trước, không được đổi cron thành 02:00 rồi coi data_interval_start/end vẫn là nửa đêm. Cần một quy tắc mapping ngày nghiệp vụ hoặc timetable phù hợp, được kiểm thử riêng. Bài giữ mốc nửa đêm để ngữ nghĩa đơn giản; việc nguồn đã hoàn tất hay chưa vẫn cần kiểm tra readiness và độ trễ ingestion.

4. DAG đầy đủ: lịch tự động dùng interval, chạy tay yêu cầu ngày rõ ràng

Tạo dags/retail_daily.py, giữ dags/retail_common.py của bài 2 vì bài này import lại hàm ingest_orders từ module không tạo DAG. Chúng ta không chép một bản logic gọi API khác vào DAG mới. Code gốc vẫn được giữ thành DAG riêng cho việc học, nhưng hãy pause retail_api_to_postgres khi thử DAG theo ngày để tránh hai DAG cùng ghi bảng raw.orders.

from datetime import timedelta
import re
import pendulum
from airflow.sdk import dag, task, get_current_context
from airflow.timetables.interval import CronDataIntervalTimetable
from retail_common import ingest_orders

ZONE = 'Asia/Ho_Chi_Minh'


def resolve_window(context):
    run = context['dag_run']
    configured = (run.conf or {}).get('process_date')
    if run.run_type == 'manual':
        if not isinstance(configured, str) or not re.fullmatch(r'\d{4}-\d{2}-\d{2}', configured):
            raise ValueError('Manual run requires process_date YYYY-MM-DD')
        start = pendulum.from_format(configured, 'YYYY-MM-DD', tz=ZONE)
        end = start.add(days=1)
    else:
        start = context.get('data_interval_start')
        end = context.get('data_interval_end')
        if start is None or end is None or start >= end:
            raise ValueError('Scheduled/backfill run requires a positive data interval')
    return start.isoformat(), end.isoformat()


@dag(
    dag_id='retail_daily_orders',
    schedule=CronDataIntervalTimetable('0 0 * * *', timezone=ZONE),
    start_date=pendulum.datetime(2026, 1, 1, tz=ZONE),
    catchup=False,
    max_active_runs=1,
    tags=['techdata', 'retail', 'lesson-3'],
)
def retail_daily_orders():
    @task(retries=1, retry_delay=timedelta(seconds=20),
          execution_timeout=timedelta(minutes=5))
    def load_day():
        start, end = resolve_window(get_current_context())
        print({'requested_from': start, 'requested_to': end})
        return ingest_orders(start, end)

    @task
    def audit_day(batch):
        from airflow.providers.postgres.hooks.postgres import PostgresHook
        connection = PostgresHook(postgres_conn_id='shop_postgres').get_conn()
        try:
            with connection.cursor() as cursor:
                cursor.execute('''
                    select count(*), coalesce(sum(amount_vnd)
                        filter (where status = 'paid'), 0)
                    from raw.orders
                    where created_at >= %s::timestamptz
                      and created_at < %s::timestamptz
                ''', (batch['from'], batch['to']))
                count, revenue = cursor.fetchone()
            print({'from': batch['from'], 'to': batch['to'],
                   'fetched_orders': batch['fetched_orders'],
                   'warehouse_orders': count, 'paid_vnd': int(revenue)})
            if count != batch['fetched_orders']:
                raise ValueError('Source batch count differs from warehouse interval')
        finally:
            connection.close()

    audit_day(load_day())

retail_daily_orders()

resolve_window có hai nhánh. Với run manual, người thao tác bắt buộc truyền process_date dạng YYYY-MM-DD; code dựng đầu ngày Việt Nam và ngày kế tiếp. Với run theo lịch hoặc backfill, code dùng data_interval_start/end của context và từ chối interval thiếu hoặc không có độ dài dương. Điều này tránh phụ thuộc việc một run manual có hoặc không có logical_date/interval trong Airflow 3.

Nhánh manual không sửa logical_date của Airflow thành process_date. Nó chỉ xác định phạm vi nghiệp vụ mà code sẽ xử lý. Vì vậy một run manual được tạo hôm nay với process_date=2026-01-01 có thể vẫn hiện logical date hoặc mốc tạo run khác trong UI. Đọc requested_from/requested_to trong log để xác nhận input thật của task; không đổi tên hai khái niệm cho nhau.

Bộ lọc thời gian được truyền vào API trước khi phân trang. Hàm ingest_orders đã có sẵn kiểm tra, upsert và transaction. Task audit_day chỉ truy vấn đúng interval rồi so số đơn nguồn với số dòng warehouse trong interval. Với contract fixture ổn định, count lệch là tín hiệu cần điều tra. Một nguồn có xóa hay chuyển created_at có thể làm phép đối soát này lỗi; lab không tự xóa dòng warehouse để “làm cho bằng nhau”.

Không có hard-code tổng tiền 200.000 vào DAG vận hành, vì ngày rỗng hoặc fixture được thêm đơn là tình huống hợp lệ. Những giá trị biết trước dùng để nghiệm thu bên ngoài DAG. Bài 4 sẽ bổ sung lớp kiểm tra chất lượng có quy tắc rõ và kết hợp dbt.

5. Chạy hai ngày cố định và kiểm tra số tiền

docker compose exec airflow-scheduler airflow dags list-import-errors
docker compose exec airflow-scheduler airflow dags pause retail_api_to_postgres
docker compose exec airflow-scheduler airflow dags unpause retail_daily_orders

docker compose exec airflow-scheduler airflow dags trigger retail_daily_orders \
  --conf '{"process_date":"2026-01-01"}'

docker compose exec airflow-scheduler airflow dags trigger retail_daily_orders \
  --conf '{"process_date":"2026-01-02"}'

Chờ run ngày đầu hoàn thành rồi nghiệm thu ngày tiếp theo nếu muốn dễ theo dõi. max_active_runs=1 hạn chế các run thông thường của DAG này chạy đồng thời; không dùng nó như khóa toàn hệ thống. Nếu hệ thống còn tự chạy một interval gần hiện tại có dữ liệu rỗng, hãy đọc đúng run manual bạn vừa tạo để không nhầm kết quả.

Nếu làm tiếp từ bài 2, cả năm đơn đã tồn tại. Chạy theo ngày vẫn có thể thành công vì upsert không tạo trùng. Nếu database mới, hai lần chạy sẽ lần lượt bổ sung đúng các đơn thuộc hai ngày. Trong cả hai trường hợp, tổng cuối với fixture gốc là 5 đơn và paid 350.000 VND.

docker compose -f docker-compose.yaml -f compose.override.yaml exec -T shop-db \
  psql -U shop -d retail -c "
    select (created_at at time zone 'Asia/Ho_Chi_Minh')::date as business_date,
           count(*) as orders,
           coalesce(sum(amount_vnd) filter (where status='paid'),0) as paid_vnd
    from raw.orders
    group by 1 order by 1;"

Thử Trigger manual không truyền process_date. Kỳ vọng load_day thất bại với thông báo yêu cầu ngày, thay vì âm thầm lấy ngày hiện tại. Truyền 2026-02-30 cũng phải lỗi vì ngày không hợp lệ; truyền giá trị có thêm giờ bị từ chối bởi định dạng. Ngày lịch đúng nhưng nguồn không có đơn sẽ tạo batch rỗng; audit ghi 0 nếu warehouse cũng không có đơn trong khoảng đó.

6. Catchup, clear task và backfill giải quyết ba nhu cầu khác nhau

catchup=False không có nghĩa task không được retry và cũng không cấm backfill. Nó tránh tự động tạo cả chuỗi khoảng đã bỏ lỡ theo cơ chế scheduler. Nếu DAG tạm dừng nhiều ngày rồi được bật lại, bạn cần quyết định dữ liệu nào phải bổ sung; không giả định Airflow đã xử lý mọi ngày chỉ vì run mới nhất thành công.

Clear đưa task về trạng thái có thể được chạy lại theo cơ chế của Airflow. Code chạy lại có thể ghi thêm, cập nhật hoặc tạo side effect nếu không thiết kế idempotent. Với DAG đang dùng conf process_date, task clear vẫn cần đúng conf và đầu vào được giữ. Với lịch tự động/backfill, phải kiểm tra interval gắn với run chứ không thay bằng ngày đang ngồi xử lý sự cố.

7. Backfill theo CLI của Airflow 3.1.7

Trước khi chạy bù, xem help của chính phiên bản đang dùng. Trong nhánh 3.1.7, CLI reference và help sử dụng --from-date, --to-date, --reprocess-behavior. Không lấy nguyên lệnh Airflow 2 hoặc một đoạn ví dụ của phiên bản khác rồi sửa mò cho hết lỗi.

docker compose exec airflow-scheduler airflow backfill create --help

docker compose exec airflow-scheduler airflow backfill create \
  --dag-id retail_daily_orders \
  --from-date '2026-01-01T00:00:00+07:00' \
  --to-date '2026-01-02T00:00:00+07:00' \
  --reprocess-behavior none \
  --max-active-runs 1 \
  --dry-run

From/to ở đây xác định phạm vi logical date mà backfill xem xét, không phải truyền thẳng một cặp bộ lọc API chung cho tất cả run. Với timetable ngày của bài, hai logical date 01/01 và 02/01 tương ứng hai interval ngày liên tiếp. Dry-run giúp xem các mốc được cân nhắc trước khi thực thi. Số run thực sự được tạo còn phụ thuộc trạng thái run đã tồn tại và lựa chọn reprocess ở thời điểm chạy thật.

docker compose exec airflow-scheduler airflow backfill create \
  --dag-id retail_daily_orders \
  --from-date '2026-01-01T00:00:00+07:00' \
  --to-date '2026-01-02T00:00:00+07:00' \
  --reprocess-behavior none \
  --max-active-runs 1

none không tạo thêm nếu đã có run cho logical date đó; failed cho phép xem xét chạy lại ngày có run lỗi; completed cho phép reprocess cả ngày đã hoàn tất theo điều kiện của Airflow. Run mới nhất còn queued/running không được tạo trùng theo các lựa chọn này. Không dùng completed cho nhiều năm chỉ vì muốn chắc chắn; hãy xem phạm vi và ảnh hưởng ghi dữ liệu trước.

Giới hạn max_active_runs của một backfill được áp dụng riêng với giới hạn DAG tương ứng, nên trong lab đặt 1 cho chính backfill và tránh chạy thêm các thao tác manual lúc đó. Đây không phải khóa ghi phân tán. Nếu có nhiều nguồn kích hoạt hoặc nhiều DAG cùng ghi, cần thêm thiết kế concurrency/locking thích hợp. Khi xử lý lịch sử lớn, còn phải đánh giá tải API, database và thời gian hoàn thành.

Để kiểm tra một ngày đã thành công có được thực thi lại hay không, xem run ID, run type và interval trong UI. Không chỉ nhìn tổng số dòng warehouse, vì upsert đúng có thể khiến số dòng giữ nguyên dù đã chạy lại. Log batch và số trang là bằng chứng bổ sung về phần việc đã thực hiện.

8. Dữ liệu đến muộn: chạy lịch không đủ để tự tìm mọi thay đổi

Giả sử sau khi xử lý ngày 01/01 xong, API mới xuất hiện đơn 1006 có created_at thuộc 01/01. Run lịch hôm nay chỉ lấy ngày hôm qua, nên sẽ không tự quay lại đầu tháng để tìm đơn đó. Airflow biết lịch, không tự biết nguồn vừa bổ sung lịch sử ở đâu.

Thêm đúng object dưới đây vào cuối mảng orders.json khi không có run đang đọc fixture. Nhớ dấu phẩy phân cách các object để JSON hợp lệ:

{"order_id":1006,"created_at":"2026-01-01T23:30:00+07:00","status":"paid","amount_vnd":30000}

Chạy lại manual với process_date=2026-01-01. Kỳ vọng ngày 01/01 có 4 đơn và paid 230.000 VND; toàn bảng có 6 đơn và paid 380.000 VND. Ngày 02/01 vẫn có 2 đơn và 150.000. Một lần chạy nữa cho ngày 01/01 không được tạo đơn thứ bảy.

Sau thử nghiệm, nếu muốn trở về fixture gốc thì chỉ xóa 1006 khỏi API là chưa đủ: upsert không xóa dòng đích. Xóa riêng đơn thử trong database lab rồi chạy lại ngày 01/01 để đối soát; lệnh dưới đây chỉ dành cho bản ghi thử 1006 bạn vừa tạo, không phải quy tắc đồng bộ xóa production:

docker compose -f docker-compose.yaml -f compose.override.yaml exec -T shop-db \
  psql -U shop -d retail -c "delete from raw.orders where order_id = 1006;"

Một lựa chọn vận hành là quét lại cửa sổ gần đây, chẳng hạn ba ngày, rồi upsert theo khóa. Cửa sổ đó phải dựa trên độ trễ nguồn và chi phí; nó không bảo đảm bắt được đơn đến trễ bốn ngày. Những khoảng ngoài cửa sổ cần backfill có chủ đích hoặc cơ chế nhận diện thay đổi tốt hơn.

Còn một trường hợp khác: đơn tạo từ tháng trước đổi pending thành paid hôm nay. Nếu API chỉ lọc created_at, quét ba ngày theo ngày tạo vẫn bỏ sót thay đổi đó. Cần API lọc updated_at với watermark/version, changelog hoặc CDC; sau đó xác định ngày báo cáo nào chịu tác động. Đến muộn của một đơn mới và cập nhật một đơn cũ không nên bị gộp thành một vấn đề giống hệt nhau.

Lab dùng created_at bất biến và trạng thái hiện tại, không phải sổ ghi nhận doanh thu kế toán theo thời điểm thanh toán. Khi nghiệp vụ yêu cầu phân tích theo ngày paid_at, hoàn tiền hoặc ghi nhận doanh thu, phải thay hợp đồng dữ liệu và test tương ứng, không chỉ đổi tên cột trong báo cáo.

9. Bài lab nghiệm thu trước khi thêm dbt

1. Giải thích khoảng [01/01 00:00, 02/01 00:00) tại Việt Nam và chuyển đúng hai mốc sang UTC.
2. Chạy hai ngày mẫu bằng process_date, đối soát lần lượt 200.000 và 150.000 VND.
3. Trigger manual thiếu ngày và với ngày không tồn tại; xác nhận lỗi rõ ràng thay vì đọc nhầm ngày.
4. Chạy lại một ngày không đổi; xác nhận số dòng và tổng tiền không nhân đôi.
5. Thực hiện backfill dry-run phạm vi hai logical date, xem các interval tương ứng và chính sách reprocess trước khi chạy thật.
6. Thêm đơn đến muộn 1006, chỉ xử lý lại ngày 01/01 và xác nhận tổng ngày tăng đúng 30.000 VND.
7. Hoàn nguyên fixture và xóa riêng bản ghi thử ở đích nếu cần; ghi rõ upsert không xử lý xóa.
8. Phân biệt cách xử lý đơn mới đến muộn với cập nhật trạng thái của một đơn tạo từ rất lâu.

Một DAG theo lịch tốt cần trả lời được ba câu: run này xử lý khoảng nào, vì sao chọn khoảng đó và làm sao biết kết quả đúng. Khi ba câu này rõ, retry, clear task và backfill trở thành thao tác có kiểm soát. Bài 4 sẽ đưa dữ liệu đã nạp qua dbt và chặn bước phục vụ báo cáo khi chất lượng chưa đạt.

Xem lại bài 2: API có phân trang và upsert PostgreSQL hoặc bài 1: kiến trúc Airflow và DAG đầu tiên nếu cần dựng lại project.

Tài liệu tham khảo

• Timetable theo khoảng dữ liệu
• Airflow 3: thay đổi mặc định và context
• Template context và data interval
• Backfill và chính sách reprocess
• CLI reference đúng nhánh 3.1.7
• PostgreSQL: date/time và timezone