Lab Airflow bằng Docker

• Tải file docker-compose.yaml chính thức (đã gồm Scheduler, Webserver, Worker, Postgres, Redis):

# tạo thư mục lab và vào đó
mkdir airflow-lab && cd airflow-lab

# tải compose chính thức (chọn đúng version Airflow 2.x)
curl -LfO 'https://airflow.apache.org/docs/apache-airflow/stable/docker-compose.yaml'

• Tạo các thư mục được mount vào container và file .env khai báo AIRFLOW_UID (để file do container ghi ra không bị sai quyền):

# thư mục cho DAG, log, plugin
mkdir -p ./dags ./logs ./plugins ./config

# ghi UID hiện tại vào .env (Linux/macOS)
echo "AIRFLOW_UID=$(id -u)" > .env

• Khởi tạo metadata DB & tài khoản admin (chạy một lần):

docker compose up airflow-init

• Khởi động toàn bộ các service ở chế độ nền:

docker compose up -d

• Mở UI tại http://localhost:8080 — đăng nhập mặc định user airflow / mật khẩu airflow.

Một pipeline ETL tối giản: đọc CSV → transform → ghi kết quả (ở đây mô phỏng, in ra số dòng). Trong thực tế bước nặng nên là một SparkSubmitOperator trỏ vào cụm Spark của bạn.

from datetime import datetime, timedelta
from airflow.decorators import dag, task

default_args = {"owner": "data-team", "retries": 2,
                "retry_delay": timedelta(minutes=5)}

@dag(
    dag_id="etl_csv_to_parquet",
    default_args=default_args,
    schedule="0 6 * * *",        # 6h sáng mỗi ngày
    start_date=datetime(2026, 1, 1),
    catchup=False,
    tags=["etl", "lab"],
)
def etl_pipeline():

    @task()
    def extract():
        import csv
        path = "/opt/airflow/dags/data/sales.csv"
        with open(path) as f:
            rows = list(csv.DictReader(f))
        return {"path": path, "count": len(rows)}

    @task()
    def transform(meta):
        # mô phỏng làm sạch: loại dòng qty âm
        print(f"Transform {meta['count']} dòng từ {meta['path']}")
        return meta["count"]

    @task()
    def load(n):
        # mô phỏng ghi Parquet ra kho dữ liệu
        print(f"Ghi {n} dòng ra Parquet (partition theo ngày)")

    load(transform(extract()))

dag = etl_pipeline()

Phiên bản trigger Spark thay cho bước nặng (thay transform/load bằng một task submit Spark):

from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator

spark_job = SparkSubmitOperator(
    task_id="spark_transform",
    application="/opt/airflow/jobs/etl_job.py",
    conn_id="spark_default",        # khai báo trong Admin -> Connections
    name="etl_csv_to_parquet",
)

• Dựng Airflow bằng Docker Compose theo các bước trên, đăng nhập UI 8080 và xem các DAG ví dụ.
• Tạo file dags/etl_daily.py: viết một DAG chạy hằng ngày (schedule="@daily", catchup=False).
• DAG gồm 3 task theo TaskFlow: extract (đọc CSV mẫu) → transform → load, truyền dữ liệu qua giá trị return (XCom).
• Thay bước nặng bằng một SparkSubmitOperator gọi pipeline Spark ghi ra Parquet có partition theo ngày; cấu hình spark_default trong Admin → Connections.
• Thêm retries=2 và một FileSensor (mode reschedule) chờ file đầu vào trước khi extract.
• Trigger DAG thủ công, xem log từng task trên UI và kiểm tra kết quả Parquet.