[DE Blog #05] Làm Chủ Workflow Orchestration: Kiến Trúc Apache Airflow, DAG Design & Nghệ Thuật Biến Đổi Dữ Liệu Với dbt

1. Bối cảnh thực tế (Context & Problem Statement)

Khi mới bắt đầu, nhiều kỹ sư thiết lập lịch chạy pipeline bằng Linux Cron Jobs. Tuy nhiên, khi hệ thống mở rộng lên hàng chục nguồn dữ liệu với hàng trăm bước xử lý, Cron Job nhanh chóng trở thành "cơn ác mộng":

Không quản lý được phụ thuộc (Dependencies): Task $B$ chỉ được chạy khi Task $A$ hoàn thành thành công. Nếu Task $A$ chạy lâu hơn bình thường, Task $B$ chạy sớm sẽ xử lý thiếu dữ liệu.
Không có cơ chế Backfill tự động: Khi logic thay đổi, làm thế nào để chạy lại dữ liệu cho 6 tháng trước một cách có kiểm soát?
Thiếu khả năng giám sát (Observability): Khó theo dõi trạng thái, thời gian chạy của từng task, không có giao diện trực quan và tự động retry khi gặp lỗi mạng.

Apache Airflow ra đời như một giải pháp tiêu chuẩn để giải quyết trọn vẹn bài toán Workflow Orchestration (Điều phối quy trình).

2. Các Khái Niệm & Cơ Chế Cốt Lõi

2.1. Kiến Trúc Nội Bộ Của Apache Airflow

Airflow được cấu thành từ 5 thành phần chính:

Airflow Webserver: Giao diện người dùng (UI) trực quan để giám sát trạng thái DAG, trigger thủ công, xem logs và cấu hình connection.
Airflow Scheduler: "Trái tim" của Airflow. Nó liên tục parse code Python của các DAG, kiểm tra điều kiện phụ thuộc và thời gian để đưa các Task sẵn sàng thực thi vào hàng đợi (Queue).
Metadata Database (PostgreSQL/MySQL): Lưu trữ toàn bộ trạng thái của DAGs, Task Instances, Variables, Connections và lịch sử chạy.
Executor: Cơ chế quyết định Task sẽ được chạy như thế nào và ở đâu (chạy tuần tự, đa luồng trên 1 máy hay phân tán trên cluster).
Workers: Các tiến trình hoặc máy chủ thực sự kéo Task từ Queue về để chạy code.

+-------------------------------------------------------------+
|                      Airflow Webserver                      |
+------------------------------+------------------------------+
                               |
                               v
+------------------+   +----------------+   +-----------------+
|   DAG Files      |-->| Airflow        |<->| Metadata DB     |
|   (Python Code)  |   | Scheduler      |   | (PostgreSQL)    |
+------------------+   +-------+--------+   +-----------------+
                               |
                               v
                       +---------------+
                       |   Executor    |
                       +-------+-------+
                               |
            +------------------+------------------+
            |                                     |
            v                                     v
    +---------------+                     +---------------+
    | Celery Worker |                     | K8s Pod Worker|
    +---------------+                     +---------------+

2.2. Các Khái Niệm Quan Trọng Trong DAG Design

DAG (Directed Acyclic Graph): Đồ thị có hướng không chu trình, đại diện cho một quy trình công việc gồm nhiều Task liên kết theo thứ tự thực hiện (task_a >> [task_b, task_c] >> task_d).
Operator vs. Sensor vs. Hook:
- Operator: Định nghĩa hành động thực tế cần làm (ví dụ: PythonOperator, BashOperator, BigQueryInsertJobOperator).
- Sensor: Một dạng Operator đặc biệt dùng để "chờ đợi" một sự kiện bên ngoài xảy ra (ví dụ: S3KeySensor chờ file upload lên S3, ExternalTaskSensor chờ DAG khác hoàn thành).
- Hook: Giao diện kết nối cấp thấp với các dịch vụ bên ngoài (PostgresHook, S3Hook) để Operator sử dụng.
XComs (Cross-Communications): Cơ chế cho phép các Task trao đổi dữ liệu nhỏ với nhau thông qua Metadata DB.

2.3. Khái Niệm Then Chốt: `logical_date` (Data Interval) & `catchup`

Data Interval: Trong Airflow, một DAG chạy tại thời điểm $T$ $T$ thực chất là để xử lý dữ liệu của khoảng thời gian trước đó $[T - \text{schedule\_interval}, T]$ $[T - schedule_interval, T]$ .
- Ví dụ: DAG chạy hàng ngày (@daily), khi kích hoạt lúc 2026-08-16 00:00:00, data_interval_start sẽ là 2026-08-15 00:00:00 và data_interval_end là 2026-08-16 00:00:00.
catchup: Nếu bạn đặt start_date là 1 năm trước và catchup=True, ngay khi bạn bật DAG, Scheduler sẽ tự động tạo và kích hoạt tất cả các Task Instances của từng ngày trong suốt 1 năm qua (Backfilling tự động).

2.4. Bộ Đôi Hoàn Hảo: Airflow + dbt (Data Build Tool)

Trong kiến trúc Modern Data Stack:

Airflow đảm nhận vai trò Orchestrator: Trích xuất dữ liệu từ các nguồn (API, CDC, S3), nạp vào Raw Layer (Bronze), sau đó gọi lệnh chạy dbt.
dbt đảm nhận vai trò Transformation Engine: Chuyển đổi dữ liệu bằng SQL thuần túy, tự động xây dựng Data Lineage, quản lý Model Versioning và chạy bộ Data Quality Tests.

3. Các Bảng Markdown So Sánh Chi Tiết

📊 Bảng 1: So Sánh Các Loại Airflow Executors

Executor	Cơ chế hoạt động	Khả năng mở rộng (Scalability)	Môi trường phù hợp
SequentialExecutor	Chạy từng task tuần tự trên 1 tiến trình (dùng SQLite)	Không thể scale ( $1$ task/lần)	Chỉ dùng để học tập, test local
LocalExecutor	Chạy đa tiến trình (multiprocessing) trên cùng 1 máy chủ	Giới hạn theo CPU/RAM của máy chủ Airflow	Môi trường Dev hoặc hệ thống tải nhỏ/vừa
CeleryExecutor	Phân tán Task vào hàng đợi Redis/RabbitMQ cho cụm Celery Workers	Rất cao (thêm/bớt worker machines dễ dàng)	Môi trường Production lớn, tải ổn định
KubernetesExecutor	Mỗi Task Instance được sinh ra như một K8s Pod riêng biệt và tự hủy khi chạy xong	Tối ưu nhất (Auto-scaling linh hoạt, cô lập tài nguyên tuyệt đối)	Hệ thống Cloud-Native, tải biến động mạnh

📊 Bảng 2: So Sánh 3 Framework Workflow Orchestration Hàng Đầu

Tiêu chí	Apache Airflow	Prefect	Dagster
Mô hình triết lý	Task-based Orchestration	Workflow-as-Code (Pythonic First)	Asset-based Orchestration (Tập trung vào Dữ liệu đầu ra)
Cộng đồng & Hệ sinh thái	Lớn nhất, hàng nghìn Providers/Plugins có sẵn	Năng động, hiện đại	Phát triển rất nhanh, đặc biệt hỗ trợ cực mạnh cho Data Assets
Độ phức tạp khi cài đặt	Cao (nhiều thành phần: Web, Scheduler, Worker, DB)	Nhẹ nhàng, dễ setup ban đầu	Trung bình
Xử lý tham số động (Dynamic)	Cải thiện từ Airflow 2.3+ (Dynamic Task Mapping)	Tự nhiên theo cấu trúc code Python	Rất linh hoạt
Độ phủ trong tuyển dụng	Phổ biến nhất trong các yêu cầu tuyển dụng Data Engineer	Tăng trưởng đều	Tăng trưởng mạnh tại các công ty công nghệ mới

📊 Bảng 3: So Sánh Các Kiểu Materialization Trong dbt

Kiểu Materialization	Cơ chế tạo trong Database	Tốc độ Build	Tốc độ Query	Use-case điển hình
View	Tạo `CREATE VIEW AS ...`	Gần như tức thì	Chậm hơn (tính toán lại mỗi lần query)	Bảng dữ liệu nhỏ, logic biến đổi nhẹ
Table	Tạo bảng vật lý `CREATE TABLE AS ...`	Lâu hơn (ghi lại toàn bộ dữ liệu)	Nhanh	Bảng biến đổi phức tạp, dữ liệu vừa phải
Incremental	Chỉ nạp/biến đổi các bản ghi mới hoặc có cập nhật	Rất nhanh cho dữ liệu lớn	Nhanh	Bảng Fact hàng trăm triệu dòng, Event logs, SCD
Ephemeral	Không tạo object trong DB, được nhúng như CTE (`WITH ...`)	Không tạo bảng	Phụ thuộc câu query cha	Các bước chuẩn hóa trung gian chỉ dùng cho 1 model

4. Best Practices & Cạm Bẫy Thực Chiến (Pro-Tips & Pitfalls)

💡 Tránh viết Heavy Code ở Top-Level của DAG File

Cạm bẫy: Viết lệnh kết nối Database, gọi API ngoài hoặc đọc file lớn trực tiếp ở ngoài hàm trong file DAG.
$\rightarrow$ Scheduler parse file DAG mỗi vài giây một lần. Nếu có code nặng ở top-level, CPU của Scheduler sẽ bị chiếm dụng 100%, gây tắc nghẽn toàn bộ hệ thống!
Giải pháp: Chỉ định nghĩa cấu trúc DAG ở top-level; toàn bộ logic xử lý dữ liệu phải đặt bên trong các hàm của Operator (hoặc chạy qua Docker/K8s/Spark Operator).

💡 Không truyền Data Payload lớn qua XCom

Cạm bẫy: Trả về một Pandas DataFrame hoặc danh sách hàng triệu phần tử từ PythonOperator vào XCom.
$\rightarrow$ XCom lưu trực tiếp vào Metadata Database (PostgreSQL), khiến Database bị phình to và sập.
Giải pháp: Ghi DataFrame ra S3/GCS dạng file Parquet, và chỉ truyền đường dẫn S3 URI (String) qua XCom!

💡 Kiểm soát Data Quality với `dbt test`

Tích hợp kiểm tra tự động ngay sau khi transform:

version: 2
models:
  - name: fct_orders
    columns:
      - name: order_id
        tests:
          - unique
          - not_null
      - name: status
        tests:
          - accepted_values:
              values: ['placed', 'shipped', 'completed', 'returned']

5. Góc Ôn Luyện Phỏng Vấn (Interview Corner)

❓ Câu hỏi 1: Giải thích ý nghĩa của `logical_date` (trước đây là `execution_date`) trong Apache Airflow. Tại sao một DAG chạy vào lúc 08:00 sáng ngày 16/08 lại có logical_date là 07:00 sáng (hoặc ngày 15/08)?

Gợi ý trả lời:
- Trong Airflow, logical_date không phải là thời điểm Task bắt đầu chạy, mà là mốc bắt đầu của khoảng thời gian dữ liệu (Data Interval) mà DAG đó chịu trách nhiệm xử lý.
- Airflow thiết kế theo tư duy xử lý dữ liệu theo chu kỳ: Để xử lý trọn vẹn dữ liệu của khoảng thời gian từ 07:00 đến 08:00, hệ thống bắt buộc phải chờ đến khi thời điểm 08:00 kết thúc thì mới có đủ 100% dữ liệu của chu kỳ đó.
- Vì vậy, tại thời điểm kích hoạt lúc 08:00, DAG sẽ chạy với data_interval_start = 07:00 và data_interval_end = 08:00 (được định danh bởi logical_date = 07:00).

❓ Câu hỏi 2: So sánh CeleryExecutor và KubernetesExecutor trong Apache Airflow. Khi nào bạn sẽ chọn KubernetesExecutor cho hệ thống Production?

Gợi ý trả lời:
- CeleryExecutor: Cần duy trì một cụm Worker Nodes chạy liên tục (Static Workers). Task được phân phối qua message queue (Redis/RabbitMQ).
  - Ưu điểm: Độ trễ khởi động task cực thấp (mili-giây).
  - Nhược điểm: Khó cô lập môi trường dependency giữa các team (Python packages conflict), lãng phí tài nguyên nếu ban đêm ít task.
- KubernetesExecutor: Mỗi khi có Task cần chạy, Scheduler gọi Kubernetes API để tạo ra một K8s Pod độc lập. Khi Task hoàn thành, Pod tự động giải phóng.
  - Ưu điểm: Cô lập hoàn toàn tài nguyên và môi trường (mỗi task có thể dùng 1 Docker image riêng); tự động co giãn tài nguyên (Auto-scaling) từ $0$ lên hàng trăm Pods, tối ưu chi phí hạ tầng.
  - Nhược điểm: Phát sinh độ trễ khởi động Pod (vài giây đến vài chục giây).
- Lựa chọn: Chọn KubernetesExecutor khi doanh nghiệp có hạ tầng Kubernetes sẵn có, nhiều nhóm cùng dùng chung hệ thống (đa dạng dependencies) và lưu lượng công việc có tính biến động theo giờ.

❓ Câu hỏi 3: dbt (data build tool) hoạt động theo cơ chế nào? Làm thế nào để cấu hình Incremental Model trong dbt một cách tối ưu?

Gợi ý trả lời:
- Cơ chế hoạt động: dbt biên dịch (compile) các file mã nguồn chứa câu lệnh SELECT kết hợp cú pháp template Jinja thành câu lệnh SQL DDL/DML phù hợp với từng Data Warehouse (BigQuery, Snowflake), sau đó thực thi trực tiếp trên database đó.
- Cấu hình Incremental Model:
```
{{
    config(
        materialized='incremental',
        unique_key='order_id',
        on_schema_change='fail'
    )
}}

SELECT *
FROM {{ ref('stg_orders') }}
{% if is_incremental() %}
    -- Chỉ lọc các bản ghi mới hơn bản ghi lớn nhất đang có trong bảng đích:
    WHERE updated_at > (SELECT MAX(updated_at) FROM {{ this }})
{% endif %}
```
- Khi chạy lần đầu (dbt run), dbt tạo toàn bộ bảng (CREATE TABLE). Ở các lần chạy tiếp theo, nhờ macro is_incremental(), dbt chỉ quét các dòng mới và tự động sinh câu lệnh MERGE dựa trên unique_key để cập nhật bảng đích.

6. Tóm Tắt Ghi Nhớ Nhanh (Key Takeaways)

Apache Airflow là công cụ điều phối (Orchestration), không phải là công cụ tính toán/xử lý dữ liệu lớn (Execution Engine).
KubernetesExecutor mang lại khả năng cô lập môi trường bằng Docker Container và tự động co giãn tài nguyên tối ưu nhất.
Luôn giữ code ở top-level của DAG file thật nhẹ nhàng để tránh gây nghẽn Airflow Scheduler.
dbt kết hợp với Airflow tạo nên chuẩn mực Modern Data Stack: Airflow trigger lịch trình và dbt chịu trách nhiệm biến đổi dữ liệu (Transformation & Testing).