Apache Airflow: Xây dựng và vận hành Data Pipeline thực tế · Bài 1/6. Mỗi sáng, cửa hàng cần một báo cáo doanh thu của ngày hôm trước. Để có báo cáo, dữ liệu phải được lấy từ hệ thống bán hàng, đưa vào database, kiểm tra rồi mới tổng hợp. Nếu bước lấy dữ liệu lỗi nhưng báo cáo vẫn chạy, con số trông có vẻ hợp lệ mà thực chất đang thiếu đơn hàng.

Airflow giúp mô tả các công việc, thứ tự phụ thuộc và trạng thái thực thi của quy trình đó. Trong bài đầu, chúng ta dựng một môi trường học bằng Docker Compose, viết một DAG nhỏ rồi đọc kết quả của chính lần chạy. Bài 2 sẽ thay dữ liệu giả trong Python bằng API và PostgreSQL; bài 3 bổ sung lịch theo ngày dữ liệu, chạy bù và xử lý dữ liệu đến muộn. Cả ba bài dùng chung một thư mục project.

Đối tượng phù hợp là người đã biết Python ở mức hàm, list/dict và vài lệnh Docker cơ bản. Bạn chưa cần biết Kubernetes, Celery hay Spark. Phiên bản thực hành được cố định ở Airflow 3.1.7; đây là phiên bản của lab, không phải tuyên bố phiên bản mới nhất. Khi dùng tài liệu, hãy chọn đúng nhánh 3.1.7 để tránh trộn cú pháp giữa các đời Airflow.

1. Bài toán bán hàng: cần điều phối trước khi cần nhiều công cụ

Giả sử cửa hàng có ba công việc: lấy đơn, tính doanh thu đơn đã thanh toán và ghi kết quả cho người xem. Nếu chạy tay, bạn tự nhớ bước nào xong rồi mới chạy bước sau. Khi số quy trình tăng, việc nhớ lịch và xử lý lần chạy lỗi trở thành công việc lặp lại. Một người nghỉ phép hoặc một máy mất mạng có thể khiến báo cáo bị bỏ quên.

Airflow ghi lại quy trình bằng mã Python. Khi đến thời điểm phù hợp, nó tạo lần chạy, xác định task đủ điều kiện, giao task để thực thi và theo dõi kết quả. SQL vẫn do database thực thi, Python vẫn làm việc của Python, còn Airflow quản lý tiến trình phối hợp. Airflow không tự hiểu “doanh thu đúng” nếu bạn chưa viết kiểm tra cho khái niệm đó.

Với một script nhỏ chạy vài phút mỗi tuần, cron có thể đủ. Airflow trở nên có ích khi có nhiều bước phụ thuộc, cần chạy lại một phần, theo dõi lịch sử hoặc nhiều người cùng vận hành. Đừng chọn Airflow chỉ để bọc một câu print; bài đầu cố ý nhỏ để học cách hệ thống hoạt động, rồi mới tăng mức thực tế.

2. DAG, task và task instance: phân biệt thiết kế với lần thực thi

DAG là bản thiết kế công việc có hướng và không có vòng phụ thuộc. “Có hướng” nghĩa là có quan hệ trước/sau. “Không có vòng” nghĩa là A không được chờ B trong khi B lại chờ A. Trong project của chúng ta, phải lấy đơn trước khi tổng hợp và phải tổng hợp trước khi báo kết quả.

Task là một đơn vị công việc được đặt tên: extract_demo, summarize hoặc report. DAG Run là một lần thực thi bản thiết kế DAG. Task Instance là lần thực thi của một task trong một DAG Run cụ thể. Cùng task summarize có thể thành công ở lần chạy hôm qua và thất bại ở lần chạy hôm nay; hai trạng thái đó không mâu thuẫn.

Một hàm Python được gắn @task sẽ trở thành định nghĩa task. Khi gọi hàm đó trong phần dựng DAG, Airflow tạo quan hệ công việc, chưa lấy đơn ngay tại thời điểm đọc file. Phần xử lý bên trong hàm sẽ chạy khi task được thực thi. Điểm này giúp tránh đặt thao tác gọi API hoặc truy vấn database vào lúc Airflow đang quét file DAG.

3. Ai đọc DAG, ai quyết định, ai thực thi?

Bản Docker Compose chính thức của lab dùng CeleryExecutor. Bạn sẽ thấy nhiều container dù chỉ viết một DAG. Chúng là các vai trò của cùng một hệ thống, không phải mỗi container tương ứng một pipeline. Một worker có thể xử lý task của nhiều DAG nếu cấu hình và tài nguyên cho phép.

Khi bạn bấm Trigger, hệ thống tạo một DAG Run. Scheduler xem task nào đã đủ điều kiện, executor phối hợp giao việc, worker thực hiện hàm Python và kết quả được phản ánh lên UI. Một task bị queued lâu thường cần kiểm tra worker và khả năng nhận việc; việc sửa câu lệnh print trong task sẽ không giải quyết một worker đang dừng.

4. Dựng môi trường học bằng Docker Compose

Dùng Linux, macOS hoặc Linux trong WSL với Docker Engine/Docker Desktop và Docker Compose v2.14 trở lên. Nên dành khoảng 8 GB RAM cho Docker để lab nhiều service thoải mái hơn; máy chỉ đủ 4 GB có thể khó chạy khi còn mở nhiều ứng dụng. Đây là gợi ý môi trường học, không phải công thức tính tài nguyên production. Lệnh bên dưới thực hiện trong shell Linux/WSL/macOS, không dán nguyên sang PowerShell.

docker --version
docker compose version
mkdir -p ~/airflow-retail
cd ~/airflow-retail
mkdir -p dags logs plugins config
curl --fail --location \
  https://airflow.apache.org/docs/apache-airflow/3.1.7/docker-compose.yaml \
  --output docker-compose.yaml

docker là engine chạy container; docker compose là plugin đọc cấu hình nhiều service. Có Docker chưa chắc đã có Compose. Nếu docker compose version báo không tìm thấy, cài Compose plugin theo hướng dẫn chính thức phù hợp hệ điều hành trước khi tiếp tục. Không đổi sang docker-compose đời cũ chỉ để câu lệnh hết báo lỗi.

Tạo .env trong thư mục project. Trên Linux/WSL, dùng UID hiện tại để hạn chế file bind mount bị tạo với chủ sở hữu không phù hợp. Chỉ dùng lệnh ghi file này khi đang tạo project mới; nếu .env đã có cấu hình, mở file và bổ sung thay vì ghi đè:

printf 'AIRFLOW_UID=%s
AIRFLOW_IMAGE_NAME=apache/airflow:3.1.7
' "$(id -u)" > .env

Với macOS, có thể dùng AIRFLOW_UID=50000 trong .env. UID là mã người dùng ở cấp hệ điều hành, không phải tài khoản đăng nhập UI. Không tự đặt UID=0 để xử lý lỗi quyền rồi mặc định chạy mọi thứ bằng root.

Mở docker-compose.yaml và chỉnh hai dòng dưới đây tại đúng vị trí: LOAD_EXAMPLES trong x-airflow-common/environment và port của service airflow-apiserver. Đây là đoạn trích để sửa, không phải toàn bộ file Compose:

AIRFLOW__CORE__LOAD_EXAMPLES: 'false'

# Inside services.airflow-apiserver.ports:
- "127.0.0.1:8080:8080"

Tắt DAG ví dụ để danh sách dễ đọc. Bind 127.0.0.1 giới hạn cổng UI trên máy đang chạy Docker; tài khoản lab không nên được phơi trực tiếp ra Internet. Nếu đang chạy Docker trên VPS, localhost của trình duyệt trên laptop không phải localhost của VPS; lab này ưu tiên máy cá nhân, còn truy cập từ xa cần cơ chế truy cập được cấu hình riêng.

docker compose config --quiet
docker compose pull
docker compose up airflow-init
docker compose up -d
docker compose ps

config --quiet kiểm tra Compose có đọc được cấu hình không; không có nghĩa DAG Python đã đúng. pull tải image trước để dễ phân biệt lỗi tải với lỗi khởi động; up vẫn có thể tải image còn thiếu. airflow-init thực hiện khởi tạo database và tài khoản lab, rồi kết thúc với mã 0 là bình thường. Lệnh up -d sau đó mới giữ các service dài hạn chạy nền. Đừng dừng sau airflow-init rồi chờ UI xuất hiện.

Chờ các service dài hạn lên trạng thái khỏe, mở http://localhost:8080. File Compose của nhánh 3.1.7 được dùng ở đây cấu hình FAB Auth Manager và tài khoản mặc định airflow/airflow cho lab; điều này khác mặc định SimpleAuthManager của Airflow 3 nói chung. Nếu bạn sửa tài khoản ở bước init, dùng tài khoản đã thiết lập. Đổi giá trị biến sau khi user đã tồn tại không đồng nghĩa mật khẩu được đổi tự động.

docker compose logs --tail=80 airflow-init
docker compose logs --tail=80 airflow-apiserver airflow-scheduler airflow-worker

Đọc tên service trong log thay vì chỉ nhìn một dòng “unhealthy”. API server lỗi có thể là thiếu RAM, database chưa sẵn sàng hoặc cấu hình sai. Nếu Docker báo permission denied khi truy cập socket, cần sửa quyền sử dụng Docker theo hệ điều hành; restart container không tự cấp quyền cho tài khoản host.

5. Hiểu đúng thư mục trên máy và trong container

Bạn viết file tại ~/airflow-retail/dags trên host. Compose gắn thư mục đó vào /opt/airflow/dags trong các container Airflow. Hai đường dẫn khác nhau nhưng trỏ tới nội dung được chia sẻ bằng bind mount. Vì vậy không cần đặt project host trong /opt; vị trí bên trái của mount và vị trí bên phải có vai trò khác nhau.

Dấu chấm trong ./dags được hiểu theo project Compose tương ứng. Chạy lệnh từ đúng thư mục ~/airflow-retail giúp tránh nhầm project. Dữ liệu metadata PostgreSQL dùng volume riêng trong Compose; nó không nằm trong thư mục dags. docker compose down dừng và xóa container/network của project nhưng giữ named volume theo mặc định. Chỉ thêm --volumes khi bạn chủ đích xóa dữ liệu lab và chấp nhận khởi tạo lại.

6. Viết DAG bán hàng đầu tiên

Tạo dags/retail_hello.py với nội dung đầy đủ sau. Ba đơn chỉ là fixture nhỏ để học, chưa phải gọi hệ thống bán hàng thật. Tổng giá trị các đơn là 250.000 VND, nhưng doanh thu paid chỉ là 200.000 VND vì đơn cancelled không được tính.

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

@dag(
    dag_id='retail_hello',
    schedule=None,
    start_date=pendulum.datetime(2026, 1, 1, tz='Asia/Ho_Chi_Minh'),
    catchup=False,
    max_active_runs=1,
    default_args={'retries': 1, 'retry_delay': timedelta(seconds=20)},
    tags=['techdata', 'retail', 'lesson-1'],
)
def retail_hello():
    @task
    def extract_demo():
        return [
            {'order_id': 1001, 'status': 'paid', 'amount_vnd': 120000},
            {'order_id': 1002, 'status': 'paid', 'amount_vnd': 80000},
            {'order_id': 1003, 'status': 'cancelled', 'amount_vnd': 50000},
        ]

    @task
    def summarize(orders):
        if len({row['order_id'] for row in orders}) != len(orders):
            raise ValueError('Duplicate order_id')
        return {
            'orders': len(orders),
            'paid_revenue_vnd': sum(
                row['amount_vnd'] for row in orders if row['status'] == 'paid'
            ),
        }

    @task
    def report(summary):
        print(f"Orders: {summary['orders']}")
        print(f"Paid revenue VND: {summary['paid_revenue_vnd']}")

    report(summarize(extract_demo()))

retail_hello()

schedule=None nghĩa là DAG chưa có lịch tự động. start_date được khai báo cố định và có múi giờ; ở bài này nó không tự tạo run vì chúng ta chạy bằng Trigger. catchup=False ghi rõ ý định không tự tạo chuỗi chạy bù khi về sau có lịch. max_active_runs=1 giới hạn run hoạt động đồng thời của DAG theo cơ chế tương ứng, không đảm bảo toàn bộ các DAG khác trong hệ thống cũng chỉ có một run.

extract_demo trả ba dictionary. summarize kiểm tra order_id không trùng rồi cộng amount_vnd cho đơn paid. report chỉ in kết quả đã tính. Khi truyền kết quả extract_demo() vào summarize(), TaskFlow xác lập phụ thuộc và trao đổi kết quả nhỏ qua XCom. Đây là lý do không cần thêm một lệnh >> riêng cho cùng quan hệ đó.

XCom phù hợp thông tin nhỏ như số dòng, ID batch hoặc đường dẫn object. Ba dictionary trong bài đầu đủ nhỏ để minh họa; đừng suy ra nên đưa hàng triệu đơn hoặc một DataFrame lớn qua XCom. Bài 2 sẽ xử lý dữ liệu trong task, lưu vào PostgreSQL và chỉ trả metadata ngắn.

Không có dòng truy vấn database hay gọi API ở cấp ngoài hàm task. File có thể được Airflow đọc nhiều lần để cập nhật định nghĩa; đặt các thao tác đó ở cấp top-level có thể gây truy cập lặp ngoài ý muốn trước khi bạn bấm Trigger.

7. Chạy DAG và đọc kết quả từ đúng task

docker compose exec airflow-scheduler airflow dags list-import-errors
docker compose exec airflow-scheduler airflow dags list

Sau khi lưu file, DAG processor cần thời gian quét và cập nhật. Nếu chưa thấy ngay, kiểm tra import errors và log trước khi restart toàn bộ stack. Trong UI, tìm retail_hello, mở chi tiết, chuyển DAG khỏi trạng thái paused rồi Trigger một lần. Chọn đúng DAG Run vừa tạo, xem Graph hoặc danh sách task và mở log của report.

Orders: 3
Paid revenue VND: 200000

Đây là kết quả kỳ vọng của fixture, không phải ảnh chụp một môi trường production. Hãy kiểm tra cả ba task thành công, không chỉ đọc một con số trên terminal. Nếu có nhiều run, đối chiếu thời gian và run ID để tránh xem log của lần cũ.

Có thể Trigger thêm lần nữa bằng CLI sau khi DAG đã được nhận diện và unpause:

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

CLI giúp gửi yêu cầu, nhưng câu lệnh trigger kết thúc không đồng nghĩa mọi task đã xong. Quay lại UI để xem tiến trình. Việc chạy thử task/DAG bằng các lệnh test là công cụ phát triển khác, không thay thế việc xác nhận scheduler và worker đang phối hợp bình thường.

8. Tạo một lỗi có chủ đích để hiểu trạng thái và retry

Trong fixture, tạm đổi order_id của dòng thứ hai từ 1002 thành 1001. Chờ UI nhận phiên bản mới rồi Trigger một run mới. extract_demo vẫn có thể thành công vì nó chỉ trả dữ liệu; summarize sẽ phát hiện trùng và raise ValueError. Đây là lỗi dữ liệu cố ý, không phải lỗi Airflow.

default_args đặt retries=1 nghĩa là tối đa một lần thử lại sau lần thử đầu, tức có thể có hai lần thực thi. Trong khoảng chờ 20 giây, summarize có thể hiện up_for_retry. Vì dữ liệu trùng chưa thay đổi nên thử lại vẫn lỗi. report mặc định cần upstream thành công nên không chạy và có thể hiện upstream_failed. Không được xem việc chưa chạy report là một lần báo cáo thành công.

Khôi phục 1002, chờ nhận lại định nghĩa rồi tạo run mới. Lần này cả ba task phải thành công. Chúng ta dùng run mới cho bài đầu để dễ phân biệt lịch sử; thao tác clear task và chạy lại phạm vi dữ liệu sẽ được giải thích cẩn thận ở các bài sau. Retry là cơ chế thử lại, không phải cơ chế sửa dữ liệu sai.

9. Bài lab nghiệm thu trước khi sang API và PostgreSQL

1. Chụp lại tên DAG và ba task của một run thành công; giải thích vì sao report phải chờ summarize.
2. Đối soát bằng tay: chỉ hai đơn paid tạo 200.000 VND; đơn cancelled không góp vào doanh thu.
3. Thực hiện thử nghiệm trùng order_id và xác định task failed đầu tiên, số lượt thử và task downstream bị chặn.
4. Khôi phục fixture, tạo run mới và kiểm tra kết quả đúng trở lại.
5. Mở file dags trên host và xác nhận nó có tại /opt/airflow/dags trong container; giải thích vì sao không cần đặt project host dưới /opt.
6. Phân biệt PostgreSQL metadata của Airflow với PostgreSQL chứa đơn hàng sẽ thêm ở bài 2.

docker compose exec airflow-worker ls /opt/airflow/dags
docker compose exec airflow-worker airflow version

Sau bài này, điều quan trọng không phải nhớ tất cả container mà là biết đi tìm đúng nơi khi có vấn đề: lỗi import xem DAG processor/import errors; queued lâu kiểm tra worker; sai số tiền đọc code và dữ liệu của task; UI không vào được kiểm tra API server và stack. Bài tiếp theo sẽ dùng cùng nền tảng để lấy API có phân trang và nạp PostgreSQL theo cách chạy lại không tạo trùng.

Tài liệu tham khảo

• Docker Compose quick-start 3.1.7
• Kiến trúc Airflow
• Task và trạng thái thực thi
• Task SDK
• Hướng dẫn cài Docker Compose