DataOps nâng cao 3 — Điều phối với Airflow & dbt

Ở bài 2 ta đã dựng lớp transform bằng dbt. Nhưng dbt tự nó chỉ là một CLI chạy khi được gọi — nó không biết khi nào nguồn đã sẵn sàng, không tự retry khi hạ tầng chập chờn, không tự chạy lại kỳ dữ liệu bị lỗi. Đó là việc của orchestrator. Bài này đi sâu vào việc dùng Airflow để điều phối một pipeline ELT production ngân hàng, giả định bạn đã nắm khái niệm cơ bản qua Airflow tổng quan.

Vì sao cần orchestrator (chứ không chạy tay)

Với một job chạy 5 phút một lần trong ngày, gõ tay dbt run là được. Vấn đề sinh ra khi pipeline lớn lên:

• Phụ thuộc phức tạp: mart_bao_cao_cuoi_ngay chỉ được build sau khi CDC từ Core Banking đã nạp xong, sau khi tỷ giá cuối ngày về, sau khi bảng giao dịch đã đóng sổ. Chuỗi phụ thuộc này là một đồ thị (DAG), không phải một danh sách tuần tự.
• Lịch & data interval: báo cáo cuối ngày phải chạy đúng sau giờ cut-off, xử lý đúng dữ liệu của ngày logic đó — không phải "lúc script tình cờ được gọi".
• Retry & recovery: mạng tới data warehouse chập một nhịp, task chết. Con người ngủ; orchestrator retry với backoff rồi mới báo động.
• Quan sát (observability): khi sếp hỏi "báo cáo hôm qua chạy chưa, mất bao lâu, có test nào fail không", bạn cần một nơi trả lời ngay — Grid/Graph view của Airflow, không phải log rải rác trên máy ai đó.
• Backfill: phòng nghiệp vụ phát hiện quy tắc tính nợ xấu tháng trước sai. Bạn cần chạy lại 30 kỳ một cách an toàn, có kiểm soát — chạy tay 30 lần là công thức của lỗi.

Orchestrator biến các mối lo trên thành cấu hình khai báo thay vì cron + script tự chế.

Thiết kế DAG cho pipeline ELT

Một pipeline báo cáo cuối ngày điển hình gồm bốn pha, ánh xạ thẳng thành các nhóm task trong DAG:

1. EL — ingest nguồn: nạp dữ liệu thô từ nguồn (CDC, file cuối ngày, API tỷ giá) vào lớp raw của warehouse. Ở đây thường là kiểm tra tính đầy đủ của nguồn (đủ file? đủ số bản ghi kỳ vọng?).
2. Transform + test (dbt): chạy dbt build — build model theo thứ tự phụ thuộc và chạy test ngay sau mỗi model (dbt build = run + test xen kẽ theo lineage).
3. Kiểm tra chất lượng chặn (quality gate): sau transform, chạy các test nghiệp vụ mức mart (đối chiếu tổng dư nợ với sổ cái, kiểm tra không có khách hàng trùng). Test fail ⇒ chặn publish.
4. Publish / thông báo: nếu tất cả xanh, "đóng dấu" mart là chính thức (swap view, ghi metadata), refresh dashboard/extract, gửi thông báo thành công. Nếu đỏ, đi nhánh cảnh báo + quarantine.

Tích hợp dbt trong Airflow — ba cách

Đây là quyết định kiến trúc quan trọng nhất. Ba lựa chọn, đánh đổi giữa đơn giản và mức độ quan sát.

Cách 1 — BashOperator gọi dbt CLI (đơn giản nhất)

Coi cả project dbt là một task duy nhất. Airflow chỉ gọi dbt build, chờ mã thoát.

from airflow.providers.standard.operators.bash import BashOperator

dbt_build = BashOperator(
    task_id="dbt_build",
    bash_command=(
        "cd /opt/dbt/ncb_analytics && "
        "dbt build --target prod --profiles-dir . "
        "--select tag:cuoi_ngay"
    ),
    env={"DBT_ENV_SECRET_PASS": "{{ conn.dwh_prod.password }}"},  # minh hoạ
)

• Ưu: cực đơn giản, một dòng, không phụ thuộc thêm.
• Nhược: cả dbt là một hộp đen. Nếu model thứ 47 fail, task đỏ nhưng bạn phải mở log dbt để biết model nào. Retry sẽ chạy lại toàn bộ (trừ khi dùng dbt build --select result:error+ thủ công). Không có lineage model trong Airflow.

Phù hợp khi project nhỏ, hoặc khi bạn đã có quan sát riêng ở tầng dbt.

Cách 2 — astronomer-cosmos (mỗi dbt model = một task Airflow)

astronomer-cosmos đọc file manifest.json (hoặc parse project) và sinh động một TaskGroup trong đó mỗi model, mỗi test là một task Airflow riêng, giữ đúng quan hệ phụ thuộc của dbt.

from cosmos import DbtTaskGroup, ProjectConfig, ProfileConfig, RenderConfig
from cosmos.constants import TestBehavior

dbt_reports = DbtTaskGroup(
    group_id="dbt_reports",
    project_config=ProjectConfig("/opt/dbt/ncb_analytics"),
    profile_config=ProfileConfig(
        profile_name="ncb_analytics",
        target_name="prod",
        profiles_yml_filepath="/opt/dbt/ncb_analytics/profiles.yml",
    ),
    render_config=RenderConfig(
        select=["tag:cuoi_ngay"],
        test_behavior=TestBehavior.AFTER_EACH,  # test ngay sau mỗi model
    ),
)

• Ưu: observability chi tiết — Graph view hiện đúng lineage dbt, biết ngay model nào chậm/fail. Retry theo model: clear đúng task lỗi và các task xuôi dòng, không chạy lại toàn bộ. Test là task riêng ⇒ dễ thấy test nào chặn.
• Nhược: thêm dependency và độ phức tạp; số task có thể rất lớn (project trăm model ⇒ trăm task) làm scheduler nặng hơn; cần chuẩn hoá cách parse (nên build sẵn manifest.json lúc CI để tránh parse project mỗi lần).

Đây là lựa chọn mặc định tốt cho pipeline production nghiêm túc — sơ đồ Mermaid ở trên chính là mô hình này.

Cách 3 — dbt Cloud API

Nếu bạn dùng dbt Cloud, Airflow chỉ kích hoạt một job đã định nghĩa trên dbt Cloud và poll trạng thái, qua provider apache-airflow-providers-dbt-cloud (DbtCloudRunJobOperator).

• Ưu: dbt Cloud lo môi trường chạy, lịch sử, docs; Airflow chỉ điều phối "khi nào" và ghép với các bước non-dbt.
• Nhược: phụ thuộc SaaS, chi phí; observability ở mức job chứ không phải model bên trong Airflow; ràng buộc mạng/an ninh khi ngân hàng hạn chế gọi ra ngoài.

Dependencies & TaskFlow API

Cách nối các pha lại với nhau. Với Airflow hiện đại, TaskFlow API (decorator @task) cho code sạch hơn nhiều so với khai báo operator thuần, và truyền dữ liệu nhỏ giữa task qua XCom một cách tự nhiên — chi tiết ở TaskFlow & XCom.

from airflow.sdk import dag, task
from airflow.providers.standard.sensors.python import PythonSensor
import pendulum

@dag(
    schedule="0 1 * * *",          # 01:00 mỗi ngày (sau cut-off)
    start_date=pendulum.datetime(2026, 1, 1, tz="Asia/Ho_Chi_Minh"),
    catchup=False,
    default_args={"retries": 3, "retry_delay": pendulum.duration(minutes=5),
                  "retry_exponential_backoff": True},
    tags=["cuoi_ngay", "bao_cao"],
)
def bao_cao_cuoi_ngay():

    @task
    def ingest_raw(logical_dt: str) -> dict:
        # nạp raw cho đúng ngày logic; trả về số bản ghi để kiểm tra
        rows = 12345  # minh hoạ
        return {"ngay": logical_dt, "rows": rows}

    @task
    def kiem_tra_dau_vao(stats: dict) -> dict:
        if stats["rows"] < 1000:
            raise ValueError(f"Nguồn thiếu dữ liệu: {stats}")
        return stats

    @task
    def publish(ok: dict):
        # swap view mart -> chính thức, refresh dashboard
        return f"published {ok['ngay']}"

    stats = ingest_raw("{{ ds }}")
    checked = kiem_tra_dau_vao(stats)
    # DbtTaskGroup (Cosmos) đặt giữa checked và publish:
    # checked >> dbt_reports >> publish(checked)

bao_cao_cuoi_ngay()

Việc truyền dữ liệu qua XCom chỉ nên dùng cho metadata nhỏ (số bản ghi, đường dẫn, ngày) — không đẩy DataFrame qua XCom; dữ liệu lớn nằm trong warehouse, XCom chỉ mang "con trỏ".

Scheduling & data-aware

Có hai kiểu kích hoạt DAG, và pipeline production thường trộn cả hai.

Lập lịch theo thời gian: schedule="0 1 * * *" — cron cổ điển, phù hợp báo cáo cuối ngày cố định giờ.

Data-aware (Dataset/Asset triggers): thay vì "chạy lúc 1 giờ", DAG hạ nguồn tự chạy khi một Asset được cập nhật. DAG ingest khi hoàn tất sẽ outlets=[asset_raw]; DAG báo cáo khai báo schedule=[asset_raw] và tự nổ khi asset đó mới. Cách này khớp phụ thuộc theo dữ liệu, tránh việc đoán mò "chắc 1 giờ thì nguồn xong rồi". Chi tiết cơ chế Assets xem thêm ở tầng scheduling của Airflow.

Sensor chờ nguồn: khi nguồn nằm ngoài Airflow (file SFTP cuối ngày, cờ trong bảng control), dùng sensor để chờ đến khi sẵn sàng:

sensor_file = PythonSensor(
    task_id="cho_file_cuoi_ngay",
    python_callable=lambda: file_da_ve(),   # trả True/False, minh hoạ
    mode="reschedule",     # nhả worker slot giữa các lần poll
    poke_interval=300,     # poll 5 phút/lần
    timeout=60 * 60 * 3,   # bỏ cuộc sau 3 giờ -> alert
)

Luôn dùng mode="reschedule" cho sensor chờ lâu để không giữ worker slot; và luôn đặt timeout để sensor không treo vô hạn.

Xử lý lỗi: retry, alert, SLA, quarantine

Production khác demo ở chỗ lỗi được coi là chuyện thường và được xử lý có kỷ luật.

• Retry + backoff: retries=3, retry_exponential_backoff=True. Phân biệt lỗi thoáng qua (timeout mạng — retry) với lỗi logic/dữ liệu (test fail — đừng retry, sẽ fail lại). Test nghiệp vụ nên retries=0.
• Alert đúng kênh: dùng on_failure_callback gửi Slack/email kèm ngữ cảnh (DAG, task, log URL, logical_date). Đừng spam mọi lần retry — chỉ báo khi hết retry.
• SLA / deadline: báo cáo cuối ngày phải xong trước 6 giờ sáng. Cấu hình cảnh báo khi trễ deadline để đội trực biết pipeline chậm dù chưa fail hẳn.
• Quarantine: khi quality gate fail, không xoá dữ liệu cũ chính thức. Mart mới giữ ở schema staging/_pending, không swap sang view chính thức; dashboard vẫn trỏ vào bản hôm trước (dữ liệu cũ đúng còn hơn dữ liệu mới sai). Nhánh lỗi ghi log, gắn nhãn kỳ đó "cần xử lý".

def bao_slack(context):
    ti = context["task_instance"]
    msg = (f"❌ {ti.dag_id}.{ti.task_id} fail cho "
           f"{context['logical_date']}\nLog: {ti.log_url}")
    # gọi SlackWebhookHook(...).send(text=msg)  # minh hoạ

Backfill & idempotency

Đây là phần quan trọng nhất cho báo cáo và cũng dễ sai nhất.

Idempotency = chạy lại một kỳ cho ra đúng một kết quả, không nhân đôi, không rác. Nguyên tắc vàng: mỗi task phải tham số hoá theo data interval ({{ ds }}, {{ data_interval_start }}) và xoá-rồi-ghi (delete-then-insert) đúng partition của kỳ đó, thay vì INSERT mù. Với dbt, dùng incremental model có unique_key và chiến lược delete+insert/merge để chạy lại kỳ cũ không tạo bản ghi trùng.

Catchup: catchup=True khiến Airflow tự sinh mọi DAG run còn thiếu từ start_date tới hiện tại. Với pipeline mới deploy trên start_date cũ, điều này có thể nổ hàng trăm run bất ngờ ⇒ mặc định để catchup=False cho DAG production, rồi backfill có chủ đích khi cần.

Backfill có kiểm soát: khi phòng nghiệp vụ báo "quy tắc tính sai từ đầu quý", chạy lại một khoảng bằng lệnh backfill của Airflow cho đúng dải logical_date. Vì task idempotent, mỗi kỳ được ghi đè sạch. Nên chạy backfill với giới hạn song song (max_active_runs) để không đè tải warehouse trong giờ làm việc.

Tham số hoá, môi trường & secret

• Tham số hoá: không hardcode ngày/đường dẫn/tên schema. Dùng Jinja template ({{ ds }}), Airflow Variables cho cấu hình, và params cho tham số theo lần chạy (ví dụ backfill một khách hàng cụ thể).
• Tách môi trường: dev / staging / prod dùng target dbt và connection khác nhau, chọn qua biến môi trường — cùng một code DAG, khác cấu hình.
• Quản lý connection/secret — KHÔNG hardcode: mật khẩu warehouse, token dbt Cloud phải nằm trong Airflow Connections/Variables (mã hoã bằng Fernet key) hoặc secrets backend ngoài (Vault, cloud secret manager). Trong code chỉ tham chiếu tên connection (conn_id), không bao giờ ghi thẳng chuỗi bí mật vào DAG file đã commit lên Git. Đây là yêu cầu tuân thủ bắt buộc trong ngân hàng.

Giám sát DAG & production ops

Điều phối tốt phải đi kèm quan sát tốt. Airflow cho Grid/Graph view, thời lượng task, tỷ lệ fail, log tập trung. Nhưng "DAG xanh" chưa đủ — cần cả observability dữ liệu (freshness, volume, phân phối) mà bài 6 series đi sâu: một DAG có thể xanh trọn vẹn nhưng vẫn xuất báo cáo sai vì nguồn im lặng thiếu dữ liệu.

Về vận hành, các chủ đề như HA scheduler, quản lý pool/queue, nâng cấp không downtime, dọn metadata database được bàn kỹ ở Airflow production ops. Tối thiểu cho pipeline ngân hàng: chạy nhiều scheduler (HA), tách pool để pipeline cuối ngày không tranh slot với job ad-hoc, và giám sát chính bản thân Airflow (heartbeat, độ trễ scheduler).

So sánh nhanh với Databricks Workflows / Dagster

Airflow thắng khi bạn cần điều phối đa hệ thống (CDC + dbt + BI + thông báo) độc lập nền tảng — đúng ca của một ngân hàng nhiều hệ thống lõi. Dagster hấp dẫn nếu bạn muốn tư duy asset/lineage ngay từ đầu. Databricks Workflows hợp lý khi kho tính toán đã ở Databricks.

Use case thực tế

Bối cảnh: pipeline báo cáo cuối ngày của NCB gồm ~120 dbt model, chạy 01:00 giờ VN, phải xong trước 06:00 để chi nhánh mở cửa dùng dashboard.

Thiết kế:

1. PythonSensor (reschedule, poke 5 phút, timeout 3 giờ) chờ cờ "đã đóng sổ" trong bảng control của Core Banking.
2. Task EL nạp raw theo {{ ds }}, trả về số bản ghi; kiem_tra_dau_vao chặn nếu < ngưỡng kỳ vọng.
3. DbtTaskGroup (Cosmos) sinh 120 model + test thành task riêng, test_behavior=AFTER_EACH, retries=2 cho model, retries=0 cho test nghiệp vụ.
4. Quality gate: test đối chiếu tổng dư nợ mart với sổ cái sai lệch > 0.01% ⇒ fail ⇒ giữ mart ở _pending, không swap; on_failure_callback bắn Slack kênh #du-lieu-truc kèm log URL.
5. Publish: swap view chính thức, refresh extract BI, gửi thông báo thành công.

Sự cố & backfill: ngày 15 phát hiện quy tắc phân loại nợ nhóm 3 sai từ ngày 1. Vì task idempotent (dbt incremental delete+insert theo unique_key + partition ngày), đội chạy backfill dải logical_date 01→14 với max_active_runs=2 ngoài giờ cao điểm. 14 kỳ được ghi đè sạch, không bản ghi trùng, dashboard tự đúng lại sau khi mỗi kỳ publish. Toàn bộ mất ~40 phút, có log kiểm toán đầy đủ — thay vì một tuần chạy tay 14 lần đầy rủi ro.

Ghi nhớ

• Cần orchestrator vì pipeline production có phụ thuộc phức tạp, lịch, retry, quan sát, backfill — không chạy tay được.
• DAG ELT chuẩn: EL → dbt build (transform+test) → quality gate chặn → publish/thông báo, có nhánh lỗi rõ ràng.
• Tích hợp dbt: BashOperator (đơn giản, hộp đen), astronomer-cosmos (mỗi model = 1 task, observability + retry theo model — nên dùng cho production), dbt Cloud API (nếu đã mua dbt Cloud).
• TaskFlow API cho code sạch; XCom chỉ mang metadata nhỏ, dữ liệu lớn ở warehouse.
• Scheduling: trộn cron và data-aware (Asset triggers); sensor chờ nguồn với mode="reschedule" và timeout bắt buộc.
• Xử lý lỗi: retry+backoff cho lỗi thoáng qua, retries=0 cho test; alert khi hết retry; theo dõi SLA; quarantine thay vì ghi đè dữ liệu đúng bằng dữ liệu sai.
• Idempotency + catchup=False mặc định + backfill có kiểm soát (max_active_runs) là chìa khoá chạy lại kỳ cũ an toàn cho báo cáo.
• Không hardcode secret: dùng Connections/Variables (Fernet) hoặc secrets backend; chỉ tham chiếu conn_id.
• DAG xanh chưa đủ — cần thêm observability dữ liệu; Airflow linh hoạt đa hệ thống hơn Databricks Workflows, còn Dagster mạnh về asset/lineage.