Series Apache Airflow thực hành · Bài 5/6. Khi API chậm hoặc database quá tải, tăng số lần retry có thể làm hệ thống còn bận hơn. Khi dữ liệu sai trạng thái, retry thêm mười lần vẫn tạo cùng một lỗi. Bài này giúp bạn chọn đúng cơ chế: lỗi nào nên thử lại, một lượt chạy được phép kéo dài bao lâu và có bao nhiêu task được cùng truy cập một tài nguyên.

Bạn sẽ tiếp tục API giả lập và PostgreSQL từ các bài trước, thêm một DAG nạp có chính sách lỗi, một DAG mô phỏng Pools và một DAG cố ý quá timeout. Mỗi tình huống có kết quả kỳ vọng để kiểm tra trong UI. Code dùng Airflow 3.1.7; đây là lab kiểm soát lỗi, chưa phải bộ cấu hình tối ưu cho mọi doanh nghiệp.

Giữ đầy đủ retail_common.py của bài 2. Pause các DAG khác đang ghi raw.orders trong lúc thực hành DAG nạp mới. DAG pool và timeout chỉ chờ bằng sleep để dễ quan sát, không gọi API hay sửa dữ liệu; không đưa các task sleep mô phỏng này vào pipeline vận hành thật.

1. Lỗi tạm thời và lỗi xác định cần được xử lý khác nhau

Không thể phân loại chính xác mọi lỗi chỉ bằng mã HTTP. Ví dụ 404 của một file vừa được tạo trong hệ thống có độ trễ có thể cần chờ theo contract riêng. Chính sách của bài cố ý nhỏ: HTTP 4xx trừ 429 được xem là không retry; lỗi 5xx, 429 và kết nối được để Airflow retry. Khi dùng API thật, phải đọc hợp đồng của nguồn và điều chỉnh.

Đừng bắt mọi exception rồi return một dictionary có status="failed". Đối với Airflow, hàm kết thúc bình thường có thể vẫn được xem là task success. Nếu một điều kiện phải chặn downstream, cần raise exception phù hợp và để trạng thái thất bại phản ánh đúng sự cố.

2. Hiểu retries, retry_delay và exponential backoff

retries=2 nghĩa là tối đa hai lần thử lại sau lần thử đầu tiên, tức có thể có ba lần thực thi. retry_delay là khoảng nghỉ cơ sở trước lượt sau, không phải thời gian tối đa của một lượt task. retry_exponential_backoff tăng khoảng chờ theo số lượt, có yếu tố điều chỉnh; không nên ghi cứng một bảng thời điểm tuyệt đối rồi kỳ vọng mọi lượt đều chờ đúng 10, 20 và 40 giây.

@task(
    retries=2,
    retry_delay=timedelta(seconds=10),
    retry_exponential_backoff=True,
    max_retry_delay=timedelta(seconds=60),
)

Đoạn trên minh họa cấu hình; DAG đầy đủ có ở phần 5. max_retry_delay giới hạn khoảng chờ theo cơ chế tương ứng, không giới hạn tổng thời gian toàn DAG. Tổng thời gian còn gồm thời gian chạy từng lượt, chờ pool, chờ worker và các bước downstream.

Retry luôn cần đi cùng khả năng chạy lại an toàn. Trong lab, raw.orders dùng khóa order_id và upsert. Nếu API đã tải xong nhưng worker mất kết nối trước khi nhận kết quả commit, lần thử sau phải có cách xử lý cùng khóa. Với thao tác gửi email, tạo hóa đơn hay trừ tiền, chỉ upsert một bảng phụ không tự làm side effect bên ngoài thành idempotent; cần idempotency key hoặc contract thích hợp.

AirflowFailException có thể kết thúc task failed ngay, bỏ các lượt retry còn lại. Chúng ta dùng nó cho payload sai hoặc 4xx theo policy của lab. AirflowSkipException lại có nghĩa task skipped; nó không phải cách đổi tên một lỗi nghiêm trọng thành trạng thái nhẹ hơn cho báo cáo đẹp.

3. Phân biệt timeout request, task và database

Một request có connect timeout 3 giây và read timeout 10 giây không đảm bảo toàn task kết thúc trong 13 giây, vì task có thể gọi nhiều trang. execution_timeout là hàng rào ở mức task. Một task bị kill cũng không phải lời hứa rằng mọi thao tác bên ngoài đã được hoàn tác; cần hiểu cách client đóng kết nối và transaction của hệ thống đích.

Với PostgreSQL, có thể áp dụng statement_timeout cho đúng session hoặc transaction thực hiện truy vấn, ví dụ SET LOCAL statement_timeout = '30s' trong một transaction. Không chạy SET LOCAL ở connection kiểm tra rồi kỳ vọng connection khác của task nạp cũng nhận cấu hình đó. Trong bài, DAG timeout mô phỏng bằng sleep để nhìn rõ giới hạn của Airflow mà không cần tạo truy vấn gây tải.

Đừng chọn timeout chỉ theo mong muốn “càng ngắn càng tốt”. Đo thời gian bình thường của task, lượng dữ liệu và cửa sổ hoàn thành báo cáo; chọn ngưỡng có khoảng dự phòng rồi theo dõi lại. Timeout quá ngắn tạo lỗi giả, còn quá dài khiến tài nguyên bị giữ lâu khi upstream đã hỏng.

4. Pools giới hạn tài nguyên dùng chung giữa các DAG

Pool là nhóm slot được gán cho các task. Ví dụ pool shop_api có một slot để các task tham gia pool không cùng gọi API bán hàng quá mức; warehouse_jobs có hai slot để mô phỏng tài nguyên xử lý. Những task không gán cùng pool không tự bị hạn chế bởi nó, và chương trình ngoài Airflow cũng không biết pool này tồn tại.

docker compose exec airflow-scheduler airflow pools set shop_api 1 \
  "Limit concurrent lab API ingestion tasks"
docker compose exec airflow-scheduler airflow pools set warehouse_jobs 2 \
  "Two weighted slots for the resource demo"
docker compose exec airflow-scheduler airflow pools list

Tạo pool trước khi Trigger DAG. Nếu pool chưa tồn tại, task có thể chờ thay vì bắt đầu chạy. Có thể kiểm tra trong Admin → Pools nếu tài khoản có quyền. Một task dùng pool_slots=2 cần đủ hai slot cùng lúc; nếu tổng pool chỉ có một slot thì task đó không thể chạy chỉ nhờ đợi lâu hơn.

Tăng worker không giải quyết một API chỉ cho một request đồng thời. Ngược lại, pool còn trống cũng không giúp task chạy nếu worker đang dừng. Khi task chờ lâu, cần xem cả pool, giới hạn DAG, queue, worker và broker. Không diễn giải mọi trạng thái queued thành “thiếu CPU”.

5. DAG nạp API có phân loại lỗi và retry

Tạo dags/retail_resilience.py. Code giữ mọi thao tác HTTP và database bên trong task hoặc hàm được task gọi. Khi lỗi không retry, message ngắn mô tả loại lỗi; không dump toàn bộ payload hoặc credential vào log.

Tạo dags/retail_errors.py để giữ policy trong module không tự tạo DAG:

def ingest_with_policy():
    import requests
    from airflow.exceptions import AirflowFailException
    from retail_common import ingest_orders

    try:
        return ingest_orders()
    except requests.HTTPError as exc:
        status = exc.response.status_code if exc.response is not None else None
        if status is not None and 400 <= status < 500 and status != 429:
            raise AirflowFailException(f'Non-retryable HTTP status: {status}') from exc
        raise
    except (requests.Timeout, requests.ConnectionError):
        raise
    except (ValueError, KeyError, TypeError) as exc:
        raise AirflowFailException('Invalid source payload; review the data contract') from exc

Sau đó tạo dags/retail_resilience.py:

from datetime import timedelta
import pendulum
from airflow.sdk import dag, task


from retail_errors import ingest_with_policy


@dag(
    dag_id='retail_resilience', schedule=None,
    start_date=pendulum.datetime(2026, 1, 1, tz='Asia/Ho_Chi_Minh'),
    catchup=False, max_active_runs=1, max_active_tasks=2,
    tags=['techdata', 'retail', 'lesson-5'],
)
def retail_resilience():
    @task(
        pool='shop_api', pool_slots=1,
        retries=2, retry_delay=timedelta(seconds=10),
        retry_exponential_backoff=True,
        max_retry_delay=timedelta(seconds=60),
        execution_timeout=timedelta(minutes=2),
    )
    def load_with_retry():
        return ingest_with_policy()

    @task(retries=0)
    def report_batch(batch):
        print({'loaded_orders': batch['fetched_orders'], 'pages': batch['pages']})

    report_batch(load_with_retry())

retail_resilience()

Hàm ingest_with_policy gọi lại ingestion của bài 2. HTTPError có response thì kiểm tra status; timeout và connection error được ném lại để Airflow xử lý theo cấu hình retry. ValueError, KeyError và TypeError của payload được chuyển thành AirflowFailException. Lỗi database khác vẫn đi theo exception gốc; muốn phân loại sâu hơn phải dựa trên driver và SQLSTATE cụ thể, tránh coi mọi OperationalError đều là tạm thời.

Bản image và ingestion của lab dùng psycopg2 cho execute_values. Nếu tự thay sang psycopg3 hoặc thay provider ngoài bộ phiên bản đã kiểm thử, cần sửa cả phần helper ghi dữ liệu tương ứng. Thay driver không chỉ là đổi một dòng import rồi mặc định transaction và helper đều giữ nguyên hành vi.

shop_api một slot giới hạn mỗi task nạp cùng pool tại một thời điểm, không giới hạn riêng từng HTTP request giữa các chương trình bất kỳ. Ở đây mỗi task chứa nhiều request phân trang và giữ slot suốt thời gian task thực thi. Nếu cần quota request/giây chính xác, phải bổ sung rate limiter phù hợp contract nguồn.

6. Hai DAG riêng để quan sát pool và timeout

Tạo dags/retail_resource_demo.py. Hai DAG này không ghi vào raw.orders:

from datetime import timedelta
import pendulum
from airflow.sdk import dag, task


@dag(
    dag_id='retail_pool_demo', schedule=None,
    start_date=pendulum.datetime(2026, 1, 1, tz='UTC'),
    catchup=False, max_active_runs=1, max_active_tasks=3,
    tags=['techdata', 'lesson-5'],
)
def retail_pool_demo():
    @task(pool='warehouse_jobs', pool_slots=1, retries=0)
    def hold_slot(label):
        import time
        print(f'Start {label}')
        time.sleep(8)
        print(f'Finish {label}')

    hold_slot.override(task_id='light_a')('light_a')
    hold_slot.override(task_id='light_b')('light_b')
    hold_slot.override(task_id='heavy', pool_slots=2)('heavy')


@dag(
    dag_id='retail_timeout_demo', schedule=None,
    start_date=pendulum.datetime(2026, 1, 1, tz='UTC'),
    catchup=False, max_active_runs=1,
    tags=['techdata', 'lesson-5'],
)
def retail_timeout_demo():
    @task(execution_timeout=timedelta(seconds=3), retries=0)
    def too_slow():
        import time
        time.sleep(15)

    too_slow()

retail_pool_demo()
retail_timeout_demo()

Trong retail_pool_demo, ba task không có dependency nên cùng đủ điều kiện. Hai task light dùng một slot mỗi task, còn heavy dùng hai slot. Với pool hai slot, có thể có hai light chạy cùng lúc hoặc một heavy chiếm đủ hai slot; không được có heavy chạy đồng thời với một light trong cùng pool theo cấu hình này.

Thứ tự bắt đầu không được cam kết bởi thứ tự các dòng Python. Scheduler có thể chọn heavy trước hoặc hai light trước tùy điều kiện và ưu tiên. Kết quả cần nghiệm thu là tổng slot sử dụng không vượt hạn mức, không phải task_a phải luôn chạy lúc giây thứ nhất.

retail_timeout_demo có một task sleep 15 giây nhưng execution_timeout 3 giây và retries=0. Kỳ vọng task thất bại vì quá thời gian. Đây là lỗi được tạo có chủ đích để học cách tìm exception và phân biệt với task đang chờ slot. Sau thực hành, pause các DAG demo để danh sách vận hành không lẫn lỗi thử nghiệm với sự cố thật.

7. Giao diện thực hành: đọc task log theo đúng lượt thử

1. Mở DAG cần kiểm tra, chọn đúng run, rồi chọn task trực tiếp lỗi.
2. Mở Logs và xem đầu tiên exception cùng thông báo; ghi lại run ID và lượt thử.
3. Chuyển sang các lượt thử khác để xem lỗi có thay đổi sau khi nguồn khôi phục không.
4. Nếu task chưa từng bắt đầu, kiểm tra pool và queue trước; không tìm một traceback chưa được tạo.

Ảnh minh họa vị trí tab Logs và trạng thái failed. Với lab của bạn, cần tìm HTTPError cho lỗi 503, message Invalid source payload cho lỗi dữ liệu, hoặc exception timeout của task mô phỏng. Không đánh dấu task success thủ công chỉ để downstream được chạy khi dữ liệu chưa đạt.

8. Diễn tập ba nhóm lỗi có thể kiểm chứng

8.1. API 503 rồi khôi phục

touch mock-api/fail.flag
docker compose exec airflow-scheduler airflow dags unpause retail_resilience
docker compose exec airflow-scheduler airflow dags trigger retail_resilience

Task load_with_retry phải lỗi HTTP 503 và chuyển sang chờ retry nếu còn lượt. Xóa đúng cờ thử nghiệm trước một lượt kế tiếp:

rm mock-api/fail.flag

Sau khi API phục hồi, lượt kế tiếp có thể thành công; nếu đã hết lượt thì tạo run mới. Metadata phải trở về năm đơn, ba trang với fixture gốc. Retry không tạo thêm bản sao order_id. Giữ lại log lần lỗi thay vì xóa run để danh sách trông sạch.

8.2. Payload sai không nên retry vô ích

Tạm sửa status của đơn 1002 thành unknown khi không có run đang đọc file, rồi Trigger. validate_order từ chối dữ liệu; AirflowFailException khiến task failed mà không dùng hết hai retry. report_batch bị chặn. Khôi phục paid, chạy lại và kiểm tra dữ liệu đích trước khi kết luận sự cố đã hết.

8.3. Tài nguyên bận và task quá thời gian

docker compose exec airflow-scheduler airflow dags unpause retail_pool_demo
docker compose exec airflow-scheduler airflow dags trigger retail_pool_demo

docker compose exec airflow-scheduler airflow dags unpause retail_timeout_demo
docker compose exec airflow-scheduler airflow dags trigger retail_timeout_demo

DAG pool phải hoàn thành khi từng task được cấp đủ slot; DAG timeout phải failed theo thiết kế. So sánh hai tình huống này trong UI: chờ tài nguyên chưa phải thực thi lỗi, còn task đã chạy rồi quá timeout có log exception. Nếu pool còn trống mà task vẫn chờ, xem trạng thái worker và giới hạn khác thay vì tăng slot ngay lập tức.

9. Checklist nghiệm thu và lựa chọn cấu hình

1. Giải thích retries=2 có tối đa ba lượt thực thi và vì sao khoảng chờ không phải tổng thời gian task.
2. Thử 503 rồi khôi phục, xác nhận có retry và không tạo trùng đơn.
3. Thử status sai, xác nhận lỗi dữ liệu kết thúc sớm và downstream không chạy.
4. Tạo pool đúng tên, quan sát hai light hoặc một heavy sử dụng tối đa hai slot.
5. Chạy timeout demo, tìm exception của đúng run và giải thích khác với chờ pool.
6. Khôi phục fixture, xóa cờ thử, pause các DAG demo sau khi hoàn thành.

Khi áp dụng vào dự án thật, bắt đầu từ giới hạn nguồn và mục tiêu hoàn thành báo cáo, đo thời gian bình thường rồi chọn timeout/retry/concurrency. Đừng dùng một cấu hình có sẵn cho mọi task. Bài cuối sẽ bổ sung kiểm thử tự động, lịch sử publish, giám sát độ trễ và quy trình khôi phục để vận hành có bằng chứng.

Xem lại bài 4: Airflow kết hợp dbt và cổng chất lượng nếu cần ôn vì sao task failed phải chặn bước publish.

Tài liệu tham khảo

• Airflow task states, timeout và special exceptions
• Pools và pool_slots
• Airflow configuration reference
• Giao diện task log
• requests timeout
• PostgreSQL statement_timeout