Airflow 5 — Executors, workers & mở rộng quy mô

Khi số DAG tăng từ vài chục lên hàng nghìn, câu hỏi không còn là "viết DAG thế nào" mà là "task chạy ở đâu, chạy bằng gì, và làm sao để không nghẽn". Đây chính là địa hạt của executor — thành phần quyết định cách Airflow biến một task đã được lên lịch thành một tiến trình thực thi thật sự.

Bài này đi sâu vào bốn executor còn lại trong Airflow 3, những gì đã bị loại bỏ so với Airflow 2, cơ chế điều khiển đồng thời, và chiến lược scale cùng cách gỡ các nút thắt phổ biến. Nếu bạn chưa nắm kiến trúc tổng thể, hãy đọc lại Tổng quan Airflow trước; còn về nơi task "làm việc" thì tham khảo Operators & Hooks.

1. Executor là gì và vì sao nó quan trọng

Trong Airflow, scheduler chịu trách nhiệm quyết định task nào đủ điều kiện chạy (đã qua dependency, đã tới lịch, còn slot). Nhưng scheduler không tự tay chạy code của bạn. Nó bàn giao task cho executor, và executor quyết định task chạy ở đâu và như thế nào:

• Chạy ngay trong tiến trình scheduler?
• Đẩy vào một hàng đợi để các worker phân tán nhặt về?
• Tạo hẳn một pod Kubernetes riêng cho từng task?
• Gửi task tới một worker ở xa qua HTTP?

Executor không chạy logic nghiệp vụ — logic đó nằm trong operator/task; executor chỉ lo cơ chế vận chuyển và thực thi. Chọn đúng executor là quyết định kiến trúc quan trọng nhất về mặt vận hành của một cụm Airflow.

Kiến trúc Airflow 3: worker không còn nối DB trực tiếp

Một thay đổi nền tảng ở Airflow 3 (so với 2) là Task Execution API. Trong Airflow 2, worker kết nối thẳng vào metadata database để đọc/ghi trạng thái task, XCom, heartbeat... Điều này gây ra hai vấn đề lớn:

1. Bảo mật: code người dùng chạy trên worker có thể import session SQLAlchemy và can thiệp trực tiếp vào DB hệ thống.
2. Scale: mỗi worker giữ một số kết nối DB → số worker lớn làm cạn connection pool của Postgres.

Airflow 3 chèn một lớp API server (FastAPI) ở giữa. Mọi tương tác lúc chạy — trạng thái, heartbeat, XCom, connection/variable — đều đi qua Task Execution API, không truy cập DB trực tiếp. Hệ quả:

• Worker không cần driver DB, không cần credential DB, chỉ cần API token.
• Worker có thể nằm ở cụm/mạng hoàn toàn khác với phần lõi Airflow → mở đường cho remote worker và cho EdgeExecutor.
• Cô lập tốt hơn: code task không "chạm" được vào metadata DB.

Đây là bối cảnh nền để hiểu vì sao các executor phân tán (Celery, Kubernetes, Edge) trong Airflow 3 "sạch" và dễ chạy remote hơn hẳn.

2. Bốn executor trong Airflow 3

LocalExecutor

Chạy task trong các tiến trình con ngay trên máy chạy scheduler. Không cần broker, không cần worker riêng, không cần hạ tầng bổ sung. Trong Airflow 3, đây là lựa chọn đơn giản nhất (vì SequentialExecutor đã bị bỏ).

• Ưu: cài đặt tối giản, độ trễ khởi động task gần như bằng 0, tuyệt vời cho dev, PoC, và cụm nhỏ/vừa.
• Nhược: giới hạn bởi tài nguyên của một máy; scheduler và task chia sẻ CPU/RAM; không scale ngang được.

CeleryExecutor

Phân tán task ra nhiều worker thông qua một message broker (thường là Redis hoặc RabbitMQ). Scheduler đẩy task vào hàng đợi; các Celery worker (có thể ở nhiều máy) nhặt task về chạy.

• Ưu: scale ngang bằng cách thêm worker; task khởi động nhanh vì worker đã "sẵn sàng" (không phải provision hạ tầng mới mỗi task); hỗ trợ queues để định tuyến task tới nhóm worker phù hợp (ví dụ: worker GPU, worker nhiều RAM).
• Nhược: cần vận hành broker + result backend; worker luôn "sống" nên tốn tài nguyên nền kể cả khi rảnh; cô lập giữa các task trên cùng worker yếu hơn Kubernetes.

Đây là lựa chọn kinh điển cho tải ổn định, throughput cao, task tương đối đồng nhất.

KubernetesExecutor

Mỗi task instance chạy trong một pod Kubernetes riêng, được tạo động qua Kubernetes API và huỷ sau khi task xong.

• Ưu: cô lập mạnh (mỗi task một pod, một môi trường); kiểm soát tài nguyên chi tiết (request/limit CPU-RAM, image tuỳ biến theo task); co giãn theo tải thật (rảnh thì gần như 0 pod) — rất hợp workload không đồng nhất và tải bùng nổ theo đợt.
• Nhược: độ trễ khởi động cao hơn (chờ pod schedule + pull image); phụ thuộc vào một cụm Kubernetes vận hành tốt.

EdgeExecutor (provider edge3)

Phân phối task tới các worker ở xa qua HTTP(s), tận dụng chính lớp Task Execution API. Đây là executor "sinh ra" từ kiến trúc decoupled của Airflow 3.

• Dùng khi: kịch bản hybrid/đa vị trí — worker chạy ở edge, on-prem, hoặc ở một đám mây khác trong khi control plane ở nơi khác; nơi mà mở kết nối DB trực tiếp là bất khả thi hoặc không an toàn.
• Đặc điểm: worker chỉ cần gọi HTTP về API server, không cần đường mạng tới metadata DB.

Bảng so sánh

Những gì Airflow 3 đã LOẠI BỎ

So với Airflow 2, hãy nhớ rõ các thay đổi sau để tránh copy cấu hình cũ:

• Bỏ SequentialExecutor — trước đây là mặc định cho SQLite/demo; nay LocalExecutor là lựa chọn đơn giản nhất.
• Bỏ DebugExecutor — cách debug đã đổi (dùng Task SDK / chạy task cục bộ).
• Bỏ hybrid CeleryKubernetesExecutor và LocalKubernetesExecutor — chúng dựa vào trường queue để chọn "sub-executor", phải viết tay từng lớp tổ hợp, khó bảo trì.

Thay cho hybrid, từ Airflow 2.10 đã có multi-executor: cấu hình nhiều executor song song và chọn executor theo từng task qua tham số executor — đa số task chạy trên Celery, vài task nặng định tuyến sang Kubernetes, không cần lớp hybrid cứng nhắc. (Multi-executor còn tương đối mới; hãy kiểm chứng trên phiên bản Airflow 3 cụ thể trước khi dùng production.)

3. Sơ đồ: CeleryExecutor hoạt động thế nào

Điểm mấu chốt (Airflow 3): các worker không nối thẳng vào Metadata DB — mọi trạng thái đi qua API server. Với KubernetesExecutor, hình dung thay "worker" bằng "một pod được tạo mới cho từng task rồi huỷ sau khi xong".

4. Điều khiển đồng thời (Concurrency)

Chọn executor mới là một nửa; nửa còn lại là giới hạn để cụm không tự bóp nghẹt mình. Airflow có nhiều tầng "van" điều tiết, từ toàn cụm xuống từng task.

Các tham số cốt lõi

• parallelism — số task tối đa chạy đồng thời trên toàn bộ cụm (per scheduler). Đây là trần cứng cấp hệ thống. Đặt 0 nghĩa là không giới hạn.
• max_active_tasks_per_dag — số task chạy đồng thời tối đa trên tất cả các run của một DAG (mặc định thường là 16). Có thể ghi đè ở cấp DAG.
• max_active_runs_per_dag — số DAG run đang active tối đa mà scheduler tạo cho một DAG. Rất quan trọng cho DAG chạy dày (ví dụ mỗi 5 phút) hoặc khi catchup/backfill.
• priority_weight — khi slot được giải phóng, task trong hàng đợi được ưu tiên chạy theo trọng số này (mặc định 1). Dùng để "chen hàng" cho task quan trọng.

Pools — hạn ngạch tài nguyên chia sẻ

Pool là cơ chế giới hạn số task đồng thời cho một nhóm task tuỳ ý, độc lập với DAG. Đây là công cụ số một để bảo vệ hệ thống bên ngoài: ví dụ database nguồn chỉ chịu được 5 kết nối song song, hay API bên thứ ba có rate limit.

Bạn tạo pool với một số slot, rồi gán task vào pool đó. Mỗi task chiếm 1 slot (hoặc nhiều hơn nếu cấu hình). Khi pool hết slot, các task còn lại xếp hàng chờ — bất kể cụm còn parallelism hay không.

# airflow.cfg (minh hoạ) — cấu hình cấp cụm
[core]
executor = CeleryExecutor
parallelism = 128
max_active_tasks_per_dag = 32
max_active_runs_per_dag = 4

# Gán task vào pool + queue + trọng số ưu tiên (minh hoạ)
extract = SomeOperator(
    task_id="extract_from_oracle",
    pool="oracle_pool",        # pool này chỉ có 5 slot
    priority_weight=10,        # ưu tiên cao
    queue="gpu",               # (chỉ với Celery) định tuyến tới nhóm worker
    executor="KubernetesExecutor",  # (multi-executor) chạy riêng trên K8s
)

Thứ tự các "van" hoạt động: một task chỉ chạy khi đồng thời thoả parallelism (toàn cụm) và max_active_tasks_per_dag (theo DAG) và còn slot trong pool của nó.

Cách tính worker / concurrency (Celery)

Một cách ước lượng thô cho CeleryExecutor:

Số task đồng thời tối đa ≈ (số worker) × (worker_concurrency)

Trong đó worker_concurrency là số task một worker chạy song song. Nếu bạn có 4 worker, mỗi worker worker_concurrency=16 → tối đa 64 task chạy cùng lúc — nhưng vẫn bị chặn bởi parallelism. Do đó nguyên tắc là: parallelism nên ≥ tổng năng lực worker, nếu không worker sẽ nhàn rỗi vì trần cụm quá thấp.

Lưu ý chọn worker_concurrency theo bản chất task: task I/O-bound (chờ mạng, chờ DB) có thể đặt cao; task CPU-bound nên đặt gần số core để tránh tranh chấp CPU.

5. Triggerer & deferrable — scale sensor rẻ hơn

Nhiều task chỉ "ngồi chờ": chờ file xuất hiện, chờ job bên ngoài xong, chờ tới giờ. Nếu mỗi task chờ chiếm một slot worker/pod, bạn lãng phí khủng khiếp.

Deferrable operators giải quyết điều này: task nhường quyền thực thi khi vào trạng thái chờ, giải phóng worker slot, và giao việc "canh chừng" cho Triggerer — một tiến trình async (asyncio) chạy hàng nghìn trigger nhẹ đồng thời trên một tiến trình duy nhất. Khi điều kiện thoả, task được đánh thức và tiếp tục.

Ở quy mô lớn, chuyển sensor sang chế độ deferrable (hoặc dùng operator hỗ trợ defer) có thể giảm mạnh số worker/pod cần thiết, vì hàng nghìn tác vụ chờ giờ nằm gọn trong Triggerer thay vì chiếm slot thực thi.

6. Nút thắt & chiến lược scale

Khi cụm "chậm", ít khi lỗi ở executor — thường là ở các nút thắt sau. Airflow 3 tách thành phần rõ ràng (scheduler, API server, DAG processor, triggerer, worker) nên có thể scale từng phần độc lập.

Scheduler throughput

Scheduler quyết định tốc độ task được lên lịch. Nếu task "đủ điều kiện" nhưng lâu mới chuyển sang running, scheduler đang là nút thắt. Cách gỡ: chạy nhiều scheduler (active-active), tinh chỉnh vòng lặp lập lịch, và giảm số DAG/task rối rắm.

DAG parsing (DAG processor)

Airflow 3 tách hẳn DAG processor thành tiến trình riêng để parse các file DAG. File DAG có top-level code nặng (gọi API, query DB lúc import) sẽ parse chậm → task bị trễ. Nguyên tắc vàng: giữ top-level code nhẹ, chỉ định nghĩa cấu trúc; việc nặng phải nằm trong task. Dọn code DAG luôn hiệu quả hơn là tăng độ song song của processor.

Kết nối Metadata DB

DB là trái tim. Ở Airflow 2, số worker lớn dễ làm cạn connection pool. Airflow 3 giảm đau nhờ Task Execution API (worker không nối DB), nhưng scheduler, API server, triggerer vẫn cần DB khoẻ. Hãy dùng Postgres được tune tốt, đặt connection pooler (PgBouncer) trước DB, và theo dõi số kết nối.

Số worker & năng lực thực thi

Chỉ khi ba tầng trên "thông" thì thêm worker/pod mới có ý nghĩa: thêm worker khi task xếp hàng lâu dù scheduler đã đẩy chúng vào queue. Với Kubernetes, đảm bảo cụm đủ node và autoscaler tốt để pod không kẹt ở Pending.

Mẹo tối ưu nhanh

• Đặt max_active_runs_per_dag hợp lý cho DAG chạy dày để tránh bùng nổ run đồng thời.
• Dùng pool cho mọi tài nguyên ngoài có giới hạn (DB nguồn, API, license).
• Chuyển sensor sang deferrable để giải phóng slot.
• Định tuyến bằng queue (Celery) hoặc multi-executor để tách task nặng khỏi task nhẹ.
• Giữ DAG file nhẹ; tránh I/O ở top level.
• Theo dõi độ trễ lập lịch (từ "đủ điều kiện" đến "running") như chỉ số sức khoẻ chính.

7. Use case thực tế

Chọn executor cho một tổ chức. Một công ty fintech chạy ~800 DAG: phần lớn là ETL SQL và gọi API nhẹ (đồng nhất, throughput cao), nhưng có một nhóm job ML huấn luyện mô hình cần image riêng và nhiều RAM (không đồng nhất). Phương án gọn: dùng CeleryExecutor làm nền cho khối ETL (worker luôn sẵn sàng, khởi động nhanh), và định tuyến các job ML sang KubernetesExecutor qua multi-executor (đặt executor="KubernetesExecutor" trên các task đó). Phần đông task hưởng độ trễ thấp của Celery, còn nhóm ML được cô lập và cấp tài nguyên riêng theo pod — không phải bơm RAM cho toàn bộ worker Celery.

Xử lý nghẽn scheduler. Team báo "task chạy trễ 10–15 phút dù cluster CPU nhàn". Chẩn đoán: đo độ trễ lập lịch → cao; kiểm tra DAG processor → thời gian parse một số file lên tới vài giây vì có requests.get(...) ở top level (chạy mỗi lần parse). Khắc phục: chuyển các lời gọi đó vào trong task, giảm số biến top-level nặng → thời gian parse giảm mạnh, độ trễ lập lịch về mức bình thường. Sau đó mới cân nhắc thêm scheduler thứ hai. Bài học: luôn loại nút thắt parsing/scheduler trước khi ném thêm worker vào.

Chi tiết vận hành, giám sát và triển khai production được nói kỹ trong Vận hành production.

Ghi nhớ

• Executor quyết định task chạy ở đâu/thế nào, không chạy logic nghiệp vụ. Airflow 3 còn 4 executor: Local, Celery, Kubernetes, Edge.
• Airflow 3 bỏ SequentialExecutor, DebugExecutor và các hybrid (CeleryKubernetes/LocalKubernetes). Thay bằng multi-executor (từ 2.10): chọn executor theo từng task.
• Kiến trúc mới: worker giao tiếp qua Task Execution API, không nối DB trực tiếp → an toàn hơn, dễ chạy remote (nền tảng của EdgeExecutor).
• Điều khiển đồng thời theo tầng: parallelism (cụm) → max_active_tasks_per_dag / max_active_runs_per_dag (DAG) → Pools (tài nguyên chia sẻ), cộng priority_weight và queue để định tuyến.
• Triggerer + deferrable giúp scale các tác vụ "chờ" mà không tốn worker.
• Khi chậm, gỡ nút thắt theo thứ tự: DAG parsing → scheduler → DB → worker. Thêm worker là bước cuối, không phải bước đầu.

Xem thêm: Tổng quan Airflow · Operators & Hooks · Vận hành production