Workflow OrchestrationWorkflow Orchestration

Thuộc bộ kiến thức Data Engineer Roadmap.

Tổng quan

Bất kỳ data pipeline nào không quá đơn giản thực chất đều là một đồ thị các bước phải chạy theo đúng thứ tự, theo một lịch trình (schedule) nhất định, và phải sống sót qua những lỗi không thể tránh khỏi của hệ thống phân tán. Extract dữ liệu từ một API, chờ nó được load vào staging table, transform nó, load vào warehouse, rồi refresh vài data mart và dashboard downstream — mỗi bước phụ thuộc vào bước trước đó, và mỗi bước có thể fail độc lập. Workflow orchestration là kỷ luật (và nhóm công cụ) quản lý việc này: biểu diễn dependency một cách tường minh, chạy task theo lịch, retry khi có lỗi tạm thời, cảnh báo (alert) con người khi có sự cố, và cho bạn một nơi duy nhất để xem pipeline đêm qua có thực sự chạy thành công hay không.

Bài viết này bao quát abstraction cốt lõi đứng sau hầu hết mọi orchestrator hiện đại — DAG — rồi đi sâu vào Apache Airflow, công cụ thống trị trong lĩnh vực này, trước khi so sánh nó với các lựa chọn mới hơn (Prefect, Dagster) và công cụ đã khởi đầu mọi thứ ở quy mô lớn (Luigi). Bài viết cũng bao gồm các khái niệm scheduling (cron, catchup/backfill, SLA), và các vấn đề vận hành về retry và idempotency giúp pipeline được orchestrate trở nên bền vững thay vì mong manh. Về các pipeline được orchestrate, xem ./08-etl-elt-and-data-pipelines.md; về việc theo dõi pipeline khi đang chạy, xem ./18-monitoring-and-observability.md; về việc test và deploy DAG code, xem ./17-cicd-for-data-engineering.md.

Vì sao cron thuần không đủ

Một dòng cron chạy script mỗi đêm hoạt động tốt — cho đến khi script đó cần gọi ba script khác theo đúng thứ tự, một trong số đó cần retry hai lần trước khi bỏ cuộc, và ai đó cần được page khi cả chuỗi mất hơn một giờ. Hãy xem một pipeline dùng cron thuần túy trông như thế nào:

0 2 * * * /usr/local/bin/extract.sh && /usr/local/bin/transform.sh && /usr/local/bin/load.sh

Nhìn có vẻ ổn cho đến khi bạn đặt những câu hỏi vận hành cơ bản:

Câu hỏi	Cron thuần	Orchestrator
Cái gì đã chạy, khi nào, và có thành công không?	Grep log file bằng tay	Web UI với lịch sử từng lần chạy
Retry chỉ bước bị fail?	Chạy lại toàn bộ chuỗi	Retry đúng task bị fail
Task B cần output của task A, không chỉ exit code	Không được mô hình hóa	Dependency tường minh + trao đổi dữ liệu (XCom, artifact)
Chạy song song hai nhánh độc lập	Cho chạy nền thủ công (&), dễ vỡ	Chạy task song song có sẵn
Backfill cho 30 ngày lịch sử bị bỏ lỡ	Tự viết vòng lặp	catchup=True / lệnh backfill
Cảnh báo on-call khi job 2 giờ sáng không xong trước 4 giờ	Script giám sát tự chế	SLA / hook cảnh báo có sẵn
Nhìn tổng thể hình dạng pipeline	Đọc shell script	Giao diện đồ thị DAG

Điều này không có nghĩa cron xấu — cron vẫn hoàn toàn ổn cho một script đơn lẻ, độc lập, idempotent, không có dependency. Vấn đề là data pipeline hiếm khi giữ được sự đơn giản đó lâu, và những kiểu lỗi ở trên cộng dồn khi số lượng script tăng lên. Orchestrator tồn tại để biến việc quản lý dependency, scheduling, retry và observability thành những mối quan tâm khai báo (declarative) hạng nhất, thay vì keo dán shell script ứng biến.

Kiến thức nền tảng

DAG: abstraction cốt lõi

Gần như mọi workflow orchestrator — Airflow, Prefect, Dagster, Luigi, thậm chí cả dbt bên trong nó — đều mô hình hóa pipeline như một Directed Acyclic Graph (DAG):

• Directed (có hướng): mỗi edge có một hướng. Một edge từ task A đến task B nghĩa là “A phải hoàn thành trước khi B bắt đầu” — không phải chiều ngược lại.
• Acyclic (không chu trình): không có chu trình (cycle). Bạn không thể để A phụ thuộc vào B trong khi B cũng (một cách gián tiếp) phụ thuộc vào A. Một task không bao giờ được phụ thuộc vào output của chính nó.
• Graph (đồ thị): node là task, edge là dependency.

Hình dạng này khớp tự nhiên với data pipeline vì các phép transform dữ liệu vốn dĩ có tính một chiều: raw data chảy vào cleaned data, cleaned data chảy vào aggregated data, aggregated data chảy vào dashboard. Không có kịch bản hợp lý nào mà dashboard cần hoàn thành trước khi raw extract chạy. Ràng buộc “acyclic” không phải là hạn chế do công cụ áp đặt — nó phản ánh một tính chất thực sự của luồng dữ liệu đúng đắn. Nếu bạn thấy mình muốn có một cycle (“task C nên trigger lại task A”), đó thường là dấu hiệu bạn thực sự muốn một DAG run mới, một vòng lặp bên trong một task đơn, hoặc một sensor chờ một điều kiện bên ngoài — chứ không phải một cycle thực sự trong dependency graph.

Một DAG chỉ định nghĩa cấu trúc (structure): những task nào tồn tại và chúng phải chạy theo thứ tự nào. Nó không nói gì về việc một task thực sự làm gì — đó là công việc của task/operator bản thân nó. Sự tách biệt này (cấu trúc vs. thực thi) là điều cho phép cùng một engine DAG có thể lên lịch những loại công việc rất khác nhau: một hàm Python, một lệnh shell, một câu SQL, một job Spark được submit, hay đơn giản là “chờ đến khi file này xuất hiện trong S3.”

Một mô hình tư duy tối giản, độc lập với bất kỳ công cụ cụ thể nào:

extract_orders     extract_customers
      \                   /
       \                 /
        v               v
         transform_and_join
                |
                v
           load_to_warehouse
                |
                v
        refresh_dashboard_mart

transform_and_join không thể bắt đầu cho đến khi cả hai task extract hoàn thành; load_to_warehouse không thể bắt đầu cho đến khi bước join xong. extract_orders và extract_customers không phụ thuộc lẫn nhau, vì vậy một orchestrator tốt sẽ chạy chúng song song.

Ôn lại cú pháp cron

Hầu hết orchestrator chấp nhận biểu thức cron chuẩn (hoặc mở rộng của nó) cho việc lên lịch. Năm trường là:

┌───────────── phút (0 - 59)
│ ┌───────────── giờ (0 - 23)
│ │ ┌───────────── ngày trong tháng (1 - 31)
│ │ │ ┌───────────── tháng (1 - 12)
│ │ │ │ ┌───────────── ngày trong tuần (0 - 6) (Chủ nhật đến Thứ bảy)
│ │ │ │ │
* * * * *

Biểu thức	Ý nghĩa
0 2 * * *	Mỗi ngày lúc 02:00
*/15 * * * *	Mỗi 15 phút
0 0 * * 0	Mỗi Chủ nhật lúc nửa đêm
0 9-17 * * 1-5	Mỗi giờ từ 9h sáng–5h chiều, Thứ 2–Thứ 6
0 0 1 * *	Nửa đêm ngày mùng 1 mỗi tháng

Airflow còn hỗ trợ thêm các preset cron như @daily, @hourly, @weekly, và — từ Airflow 2.4+ — Datasets và Timetables cho lịch trình dựa trên sự kiện (event-driven) và không theo cron (ví dụ: “chạy bất cứ khi nào dataset upstream này được cập nhật,” hoặc “chạy vào ngày làm việc cuối cùng của tháng”).

Catchup và backfill

Một điểm tinh tế khiến gần như ai mới dùng Airflow cũng vấp phải: một DAG được lên lịch chạy hàng ngày bắt đầu từ 2026-01-01 không chạy vào 2026-01-02 cho kỳ kết thúc 2026-01-02 — nó chạy sau khi 2026-01-02 00:00 kết thúc, cho khoảng thời gian [2026-01-01, 2026-01-02). Đây là mô hình data interval: mỗi lần chạy đại diện cho dữ liệu được tạo ra trong một kỳ đã hoàn tất, không phải “bây giờ.”

• catchup: nếu start_date của một DAG nằm trong quá khứ và catchup=True (mặc định lịch sử của Airflow), scheduler sẽ tự động tạo và chạy một DAG run cho mỗi khoảng thời gian bị bỏ lỡ giữa start_date và hiện tại, theo thứ tự. Deploy một DAG chạy hàng ngày với start_date cách đây ba tháng, và Airflow sẽ khởi động ~90 lần chạy backfill ngay khi nó được bật lên — thường không phải điều bạn muốn với một pipeline hoàn toàn mới chỉ quan tâm đến dữ liệu từ giờ trở đi. Hầu hết các team đặt tường minh catchup=False trừ khi backfill lịch sử thực sự được dự định.
• Backfill: hành động có chủ đích, thường là thủ công, chạy một DAG cho một khoảng ngày trong quá khứ — thường vì logic thay đổi và output lịch sử cần được tính lại, hoặc vì một pipeline mới cần populate lịch sử. Airflow hỗ trợ việc này qua airflow dags backfill -s 2026-01-01 -e 2026-01-31 my_dag, về bản chất chỉ là catchup được trigger theo yêu cầu cho một khoảng giới hạn.

Backfill cũng là nơi idempotency không còn là tùy chọn: chạy lại một task cho 2026-01-01 phải cho ra kết quả giống nhau dù đó là lần chạy đầu tiên hay lần thứ năm, nếu không backfill sẽ âm thầm làm hỏng dữ liệu (đếm trùng dòng, insert trùng lặp). Đây chính là nguyên tắc idempotency đã đề cập trong ./08-etl-elt-and-data-pipelines.md — tầng orchestration chính là nơi nguyên tắc này được kiểm chứng, vì retry và backfill đều là “chạy lại task này cho cùng một kỳ logic.”

SLA và cảnh báo

SLA theo nghĩa của Airflow là một kỳ vọng dựa trên thời gian: “task này nên hoàn thành trong X phút/giờ kể từ khi DAG run bắt đầu.” Khi một task vượt quá SLA, Airflow không kill nó — nó ghi log SLA miss và (nếu được cấu hình) kích hoạt một callback, thường tới Slack, email, hoặc PagerDuty. Điều này khác với việc task fail: một task có thể thành công nhưng vẫn bị miss SLA, và một task có thể fail rất lâu trước khi chạm đến SLA. Cả hai đều là tín hiệu hữu ích — failure cho bạn biết có gì đó đã hỏng, SLA miss cho bạn biết có gì đó đang xuống cấp (một bảng tăng trưởng chậm, một worker pool bão hòa) trước khi nó trở thành một failure thực sự.

Khái niệm chính

Kiến trúc Apache Airflow

Airflow, ban đầu được xây dựng tại Airbnb và giờ là một Apache top-level project, là orchestrator tiêu chuẩn thực tế (de facto) cho data pipeline. Kiến trúc của nó có bốn thành phần cốt lõi:

Thành phần	Vai trò
Scheduler	Trái tim của Airflow. Parse các file DAG, đánh giá lịch trình, và quyết định task instance nào sẵn sàng chạy, đưa chúng vào hàng đợi để thực thi.
Executor	Quyết định tasks thực sự chạy như thế nào và ở đâu — trong cùng tiến trình (SequentialExecutor, chỉ dùng dev), song song cục bộ nhiều tiến trình (LocalExecutor), hoặc phân tán trên một fleet worker (CeleryExecutor, KubernetesExecutor).
Workers	Các tiến trình thực sự thực thi logic task (liên quan đến Celery/Kubernetes executor — mỗi task có thể chạy trong worker/pod riêng).
Webserver	Phục vụ UI: giao diện graph/grid của DAG, log, xu hướng thời lượng task, điều khiển trigger/retry thủ công.
Metadata database	Cơ sở dữ liệu Postgres/MySQL lưu lịch sử DAG run, trạng thái task, connections, variables, XComs. Đây là nguồn sự thật (source of truth) mà cả scheduler và webserver đều đọc từ đó.

Một DAG được định nghĩa dưới dạng Python code — không phải config YAML, không phải canvas kéo-thả. Đây là một trong những quyết định thiết kế đặc trưng của Airflow: vì DAG chỉ là Python, bạn có toàn bộ sức mạnh của ngôn ngữ (vòng lặp để sinh task động, import, điều kiện, thư viện dùng chung) miễn phí, đổi lại là các file DAG cần được scheduler parse trong mỗi chu kỳ parse, đây là nguồn phổ biến gây chậm scheduler nếu file DAG làm việc nặng lúc import (ví dụ: truy vấn database để quyết định tạo task nào).

Một DAG Airflow thực tế

DAG dưới đây extract orders và customers, join/transform chúng, rồi load kết quả vào warehouse — phản ánh mô hình extract → transform → load từ ./08-etl-elt-and-data-pipelines.md, giờ được biểu diễn như một pipeline được orchestrate, có lịch trình, và có thể retry.

from datetime import datetime, timedelta

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
from airflow.sensors.filesystem import FileSensor

default_args = {
    "owner": "data-engineering",
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
    "retry_exponential_backoff": True,
    "max_retry_delay": timedelta(minutes=30),
    "email_on_failure": True,
    "email": ["data-eng-alerts@example.com"],
}

with DAG(
    dag_id="daily_orders_pipeline",
    description="Extract, transform, and load daily order data into the warehouse",
    default_args=default_args,
    schedule="0 2 * * *",       # every day at 02:00
    start_date=datetime(2026, 1, 1),
    catchup=False,              # do not backfill historical runs on deploy
    max_active_runs=1,          # don't let overlapping runs step on each other
    sla_miss_callback=None,     # plug in a Slack/PagerDuty callback here
    tags=["orders", "warehouse"],
) as dag:

    # Wait until the upstream export file has landed before doing anything else.
    wait_for_export = FileSensor(
        task_id="wait_for_export_file",
        filepath="/data/exports/orders_{{ ds_nodash }}.csv",
        poke_interval=60,
        timeout=60 * 60,        # give up after 1 hour
        mode="reschedule",      # free the worker slot between pokes
    )

    def extract_orders(**context):
        """Pull raw orders for this DAG run's data interval and stage them."""
        import pandas as pd

        ds = context["ds"]  # logical date, e.g. "2026-01-01"
        df = pd.read_csv(f"/data/exports/orders_{context['ds_nodash']}.csv")
        df.to_parquet(f"/data/staging/orders_{ds}.parquet", index=False)
        # Push a small summary to XCom for downstream tasks/observability
        context["ti"].xcom_push(key="row_count", value=len(df))

    extract = PythonOperator(
        task_id="extract_orders",
        python_callable=extract_orders,
    )

    def transform_orders(**context):
        """Clean and enrich the staged orders (idempotent: overwrite, not append)."""
        import pandas as pd

        ds = context["ds"]
        df = pd.read_parquet(f"/data/staging/orders_{ds}.parquet")
        df = df.dropna(subset=["order_id", "customer_id"])
        df["order_total"] = df["quantity"] * df["unit_price"]
        df.to_parquet(f"/data/transformed/orders_{ds}.parquet", index=False)

    transform = PythonOperator(
        task_id="transform_orders",
        python_callable=transform_orders,
    )

    # Idempotent load: DELETE + INSERT for this partition, never a blind append,
    # so retries and backfills for the same ds never double-count rows.
    load = BashOperator(
        task_id="load_orders_to_warehouse",
        bash_command=(
            "python /opt/pipelines/load_orders.py "
            "--input /data/transformed/orders_{{ ds }}.parquet "
            "--partition {{ ds }} "
            "--mode upsert"
        ),
    )

    # Trigger the dbt models that depend on the raw orders table.
    run_dbt_models = BashOperator(
        task_id="run_dbt_transform_models",
        bash_command=(
            "cd /opt/dbt_project && "
            "dbt run --select tag:orders --target prod "
            "--vars '{\"run_date\": \"{{ ds }}\"}'"
        ),
    )

    # Structure: wait -> extract -> transform -> load -> dbt models
    wait_for_export >> extract >> transform >> load >> run_dbt_models

Vài điểm đáng chú ý trong ví dụ này:

• Toán tử >> thiết lập dependency: a >> b nghĩa là “b chạy sau a.” Bạn cũng có thể viết a.set_downstream(b), hoặc xây các hình dạng fan-out/fan-in bằng list: extract >> [transform_a, transform_b] >> merge.
• Operator vs. DAG: PythonOperator, BashOperator, và FileSensor đều là các operator — mỗi cái là một template cho một loại task cụ thể. Đối tượng DAG chỉ nối chúng lại với nhau; nó không biết và không quan tâm PythonOperator thực sự thực thi cái gì bên trong. Đây chính là sự tách “DAG định nghĩa cấu trúc, task thực thi công việc” đã nhắc ở trên — thay BashOperator bằng KubernetesPodOperator thay đổi nơi và cách bước load chạy mà không đụng gì đến dependency graph.
• Sensor (FileSensor ở đây) là một nhóm operator đặc biệt chờ một điều kiện — một file xuất hiện, một partition tồn tại, một DAG bên ngoài kết thúc — thay vì trực tiếp làm việc. mode="reschedule" giải phóng worker slot giữa các lần kiểm tra thay vì giữ nó suốt, điều này quan trọng ở quy mô lớn (một sensor ở mode poke block trong một giờ sẽ giữ một worker suốt cả giờ đó).
• retries + retry_exponential_backoff thiết lập retry tự động với backoff ở mức task — xem phần Idempotency và Retry bên dưới để hiểu vì sao điều này an toàn ở đây.
• catchup=False và max_active_runs=1 là hai trong số những cấu hình bị quên phổ biến nhất gây ra sự cố thực sự: quên cái đầu tiên khiến backfill bất ngờ xảy ra khi deploy; quên cái thứ hai cho phép một lần chạy chậm chồng lấn với lần chạy theo lịch tiếp theo và làm hỏng trạng thái dùng chung.

DAG nội bộ của dbt, được orchestrate bên trong Airflow

./08-etl-elt-and-data-pipelines.md đã đề cập dbt như tầng transformation trong ELT. Từ góc độ orchestration, điều quan trọng là dbt có DAG riêng của nó, được xây dựng tự động từ các lệnh gọi ref() giữa các model:

-- models/staging/stg_orders.sql
select * from {{ source('raw', 'orders') }}

-- models/marts/fct_orders.sql
select
    o.order_id,
    o.customer_id,
    o.order_total
from {{ ref('stg_orders') }} o   -- this ref() call creates a DAG edge

dbt run (hoặc dbt build) tự giải quyết dependency graph này và thực thi các model theo đúng thứ tự, chạy song song khi có thể (dbt run --select ... --threads 4) — về bản chất, dbt là một orchestrator thu nhỏ của riêng nó, giới hạn trong phạm vi các phép transform SQL. Điều này có nghĩa một data platform thường có hai DAG lồng vào nhau: DAG của Airflow quản lý toàn bộ pipeline (extract → load raw data → chạy dbt → refresh BI extract), trong khi DAG nội bộ của dbt quản lý thứ tự dependency chi tiết bên trong một bước “chạy dbt” duy nhất. Task run_dbt_transform_models trong ví dụ trên chính xác là như vậy: từ góc nhìn của Airflow, đó là một task nguyên tử duy nhất, nhưng bên trong nó tỏa ra hàng chục lần chạy model theo thứ tự. Một số team đi xa hơn và dùng dbt-airflow hoặc astronomer-cosmos để “mở” DAG của dbt thành các task Airflow riêng lẻ (mỗi model một task) để retry và quan sát chi tiết hơn — đổi lại là một đồ thị DAG Airflow lớn hơn và phức tạp hơn nhiều.

Prefect, Dagster, và Luigi

Airflow chiếm ưu thế nhưng không phải là lựa chọn duy nhất, và các lựa chọn thay thế tồn tại phần lớn như phản ứng trước những điểm đau cụ thể của Airflow — việc parse file DAG nặng nề, dynamic workflow gượng gạo, và đường cong học tập vận hành dốc.

Luigi (Spotify mở mã nguồn năm 2012) là một trong những công cụ workflow Python phổ biến đầu tiên, ra đời trước cả Airflow. Nó mô hình hóa pipeline dưới dạng các lớp Task với các phương thức requires(), output(), và run(), và dùng cách kiểm tra “target đã tồn tại chưa” (thường là một file) cho idempotency: nếu output target đã tồn tại, task được coi là xong và bị bỏ qua. Luigi không có scheduler tích hợp riêng (thường được cron điều khiển) và web UI mỏng hơn nhiều so với Airflow. Ngày nay nó hiếm khi được chọn cho dự án mới nhưng vẫn chạy production tại nhiều công ty đã áp dụng nó cách đây một thập kỷ, và mô hình dependency dựa trên requires() của nó đã ảnh hưởng trực tiếp đến các công cụ ra đời sau.

Prefect định vị mình là một lựa chọn Pythonic hơn, ít boilerplate hơn: một hàm Python thuần được decorate bằng @task, ghép vào bên trong một hàm được decorate bằng @flow, trở thành một workflow được orchestrate — không có đối tượng DAG, không có operator, dependency được suy ra từ cách bạn gọi hàm và truyền kết quả của chúng cho nhau:

from prefect import flow, task

@task(retries=3, retry_delay_seconds=30)
def extract_orders(date: str) -> list[dict]:
    ...

@task
def transform(orders: list[dict]) -> list[dict]:
    ...

@task
def load(rows: list[dict]) -> None:
    ...

@flow(name="daily-orders-pipeline")
def daily_orders_pipeline(date: str):
    orders = extract_orders(date)
    clean = transform(orders)
    load(clean)

Tính năng chủ đạo khác của Prefect là mô hình thực thi hybrid: Prefect Cloud (hoặc Prefect server tự host) xử lý orchestration/scheduling/state, trong khi code task thực sự chạy hoàn toàn trên hạ tầng riêng của người dùng thông qua các agent nhẹ — code và dữ liệu không bao giờ phải rời khỏi mạng của bạn để control plane điều phối lần chạy. Prefect cũng hỗ trợ dynamic workflow hoàn toàn một cách tự nhiên hơn Airflow — các task graph phân nhánh dựa trên giá trị runtime (ví dụ: “tạo một task cho mỗi file tìm thấy trong thư mục này, và chúng ta không biết có bao nhiêu file cho đến runtime”) mà không cần các cách lách luật mà Airflow trước đây từng cần (dynamic task mapping được thêm sau này ở Airflow 2.3+ nhưng vốn là tính năng gốc của Prefect từ trước).

Dagster đi theo một quan điểm mạnh mẽ hơn nữa: nó được xây dựng xoay quanh software-defined assets thay vì task. Thay vì mô tả “chạy script này, rồi chạy script kia,” bạn khai báo “bảng/dataframe/model này là một asset, và nó phụ thuộc vào những asset khác này” — Dagster sau đó tự tìm ra thứ tự thực thi từ asset graph, và coi việc materialize mỗi asset (kèm metadata, freshness, và data quality check) là đơn vị quan sát chính, không chỉ đơn thuần là “task có exit 0 hay không.” Góc nhìn lấy asset làm trung tâm này khớp một cách bất thường tốt với một data stack hiện đại lấy warehouse làm trung tâm, nơi những thứ người ta thực sự quan tâm là các bảng và dashboard, không phải các script.

Bảng so sánh orchestrator

	Airflow	Prefect	Dagster	Luigi
Abstraction chính	DAG các task	Flow các hàm Python	Đồ thị software-defined asset	Task với requires()/output()
Ngôn ngữ	Python (DAG-as-code)	Python (hàm thuần + decorator)	Python (decorator asset/op)	Python
Mô hình thực thi	Scheduler + executor + worker (Celery/Kubernetes/Local)	Hybrid: control plane cloud/server, thực thi ở bất kỳ đâu qua agent	Dagster daemon + executor cấu hình được (in-process, multiprocess, Celery, K8s)	Không có scheduler tích hợp; thường do cron điều khiển, có tiến trình scheduler trung tâm để giới hạn concurrency
Dynamic workflow	Lịch sử khá cứng nhắc; dynamic task mapping thêm ở 2.3+	Native, hạng nhất	Hỗ trợ qua dynamic output	Hạn chế
Mô hình idempotency	Theo quy ước (bạn tự thiết kế)	Theo quy ước (bạn tự thiết kế)	Theo quy ước, được củng cố bởi việc theo dõi asset materialization	Tích hợp sẵn qua kiểm tra target tồn tại của output()
Độ chín / hệ sinh thái	Rất chín, hệ sinh thái plugin khổng lồ (provider cho mọi cloud, DB, công cụ SaaS)	Chín, hệ sinh thái nhỏ hơn nhưng phát triển nhanh	Chín, tích hợp dbt mạnh, hệ sinh thái nhỏ hơn Airflow	Chín nhưng phần lớn đã lỗi thời; phát triển tiếp tục thấp
Trải nghiệm dev cục bộ	Lịch sử khá cồng kềnh (cần scheduler + webserver + DB chạy)	Nhẹ nhàng; một flow chỉ là một script Python	Tốt; asset graph nhìn thấy được mà không cần deploy đầy đủ	Đơn giản nhưng công cụ tối thiểu
Phù hợp nhất	Pipeline rộng, đa dạng qua nhiều hệ thống; team muốn hệ sinh thái và cộng đồng lớn nhất	Team muốn workflow Pythonic, động, và ít boilerplate orchestration hơn	Team lấy asset/chất lượng dữ liệu làm trung tâm, đặc biệt cùng với dbt	Hệ thống legacy; hiếm khi được chọn cho công việc mới

Không công cụ nào trong số này giải quyết vấn đề scheduling/dependency nền tảng theo cách khác biệt về nguyên tắc — tất cả vẫn tạo ra thứ tương đương với một DAG bên dưới. Sự khác biệt nằm ở trải nghiệm phát triển, mức độ động của đồ thị, gánh nặng vận hành của control plane, và mức độ gắn chặt của công cụ với một mô hình tư duy cụ thể (task vs. asset).

Idempotency và retry ở tầng orchestration

Retry là tuyến phòng thủ đầu tiên của orchestrator trước những lỗi tạm thời (transient) — một trục trặc mạng khi gọi API, một kết nối warehouse timeout do tải cao, một worker mất kết nối trong chốc lát. Cấu hình retry rất rẻ:

default_args = {
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
    "retry_exponential_backoff": True,   # 5m, 10m, 20m, ... instead of flat 5m
    "max_retry_delay": timedelta(minutes=30),
}

Nhưng retry chỉ an toàn nếu task là idempotent: chạy nó hai lần cho cùng một kỳ logic phải cho ra cùng một trạng thái cuối cùng như chạy một lần. Đây chính là nguyên tắc từ ./08-etl-elt-and-data-pipelines.md, và tầng orchestration chính là nơi nguyên tắc này thực sự được kiểm chứng, vì retry (và backfill) đều, về mặt cơ chế, là “chạy lại task này cho một kỳ mà nó có thể đã chạy một phần rồi.” Cụ thể:

• Việc load nên là DELETE + INSERT hoặc MERGE/upsert trên một khóa partition, không bao giờ là INSERT/append mù quáng. Trong DAG ví dụ, load_orders_to_warehouse chạy ở --mode upsert trên partition {{ ds }} cụ thể để một retry sau lỗi một phần không tạo ra dòng trùng lặp.
• Các bước extract ghi file nên ghi đè lên đường dẫn target một cách xác định (deterministic) (orders_{ds}.parquet, không phải orders_{run_timestamp}.parquet) để retry thay thế chứ không nhân đôi dữ liệu đã staging.
• Các side effect không có “hoàn tác” tự nhiên (gửi email, gọi một API bên thứ ba không idempotent, tính phí thanh toán) cần logic khử trùng lặp tường minh — một idempotency key, một kiểm tra “tôi đã gửi cái này chưa” — vì orchestrator retry task sẽ gọi lại side effect đó y hệt.
• Retry nên được dành cho những lỗi có khả năng là tạm thời. Một task fail vì bug trong logic transform sẽ fail y hệt ở mọi lần retry, đốt hết ngân sách retry và trì hoãn cảnh báo failure — đây là điều mà exponential backoff giảm nhẹ một phần (tốn ít thời gian thực hơn để phát hiện retry sẽ không giúp ích) nhưng không giải quyết triệt để; một số team chủ động phân nhánh hành vi retry theo loại exception (retry với ConnectionError, fail nhanh với lỗi validation dữ liệu).

Giám sát sức khỏe pipeline, sơ lược

Orchestrator là nơi tự nhiên để quan sát sức khỏe pipeline vì nó đã theo dõi mọi lần chạy và trạng thái, thời lượng, kết quả của mọi task. Ít nhất, hầu hết các team thiết lập:

• Cảnh báo failure — thông báo Slack hoặc PagerDuty ngay khi một task hết retry và fail, lý tưởng là kèm link trực tiếp đến log của task bị fail.
• Cảnh báo SLA miss — một tín hiệu nhẹ nhàng hơn failure, đánh dấu các task hoặc DAG đang có xu hướng chậm trước khi chúng trở thành trễ hẳn.
• Xu hướng thời lượng task — UI của webserver (hoặc metric xuất ra Prometheus/Datadog) cho thấy một task từng mất 5 phút giờ mất 45 phút, thường là dấu hiệu báo trước một failure hoặc timeout hẳn.

Đây chỉ là lát cắt tầng orchestration của observability — data quality check, giám sát độ tươi (freshness) của dữ liệu, và xử lý sự cố đầy đủ được bao quát sâu hơn trong ./18-monitoring-and-observability.md. Việc deploy và test chính DAG code (lint DAG, chạy dbt test trong CI, deploy staging) được bao quát trong ./17-cicd-for-data-engineering.md.

Best Practices

• Thiết kế mọi task idempotent trước khi lo về retry. Retry và backfill chỉ an toàn khi đặt trên nền task idempotent; cấu hình retries=3 cho một bước load không idempotent là một bug làm hỏng dữ liệu đang chờ xảy ra, không phải một tính năng bền vững.
• Đặt catchup=False tường minh trừ khi backfill-khi-deploy thực sự được dự định. Đây là một trong những sự cố “vì sao 90 DAG run vừa chạy” phổ biến nhất với các team mới dùng Airflow.
• Giữ cho file DAG rẻ khi parse. Scheduler parse lại mọi file DAG theo một chu kỳ cố định; thực hiện truy vấn database, gọi API, hoặc tính toán nặng lúc định nghĩa DAG (bên ngoài các hàm task) làm chậm toàn bộ scheduler, không chỉ một DAG.
• Ưu tiên các task nhỏ, đơn mục đích thay vì một task khổng lồ. Một DAG với extract, transform, và load là các task riêng biệt cho phép bạn retry (và quan sát) đúng bước bị fail, thay vì chạy lại toàn bộ pipeline từ đầu.
• Dùng sensor một cách tiết chế và với mode="reschedule". Một sensor ở mode poke giữ một worker slot suốt thời gian chờ, điều này có thể âm thầm làm cạn kiệt năng lực worker khi số DAG đang chờ tăng lên.
• Đặt max_active_runs một cách có chủ đích. Không có nó, một lần chạy chậm có thể chồng lấn với lần chạy theo lịch tiếp theo, và hai lần chạy cùng ghi vào một partition đồng thời là nguồn phổ biến gây hỏng dữ liệu tinh vi.
• Truyền tham chiếu dữ liệu, không phải payload lớn, qua XCom (hoặc tương đương của Prefect/Dagster). XCom dành cho metadata nhỏ (số dòng, đường dẫn file, cờ trạng thái) — đẩy cả một dataframe đầy đủ qua nó sẽ làm phình metadata database và làm chậm scheduler.
• Version-control DAG và deploy chúng qua CI, như bất kỳ code nào khác. Đối xử với một thay đổi DAG với cùng mức độ nghiêm ngặt review như một thay đổi code ứng dụng — một DAG tồi có thể âm thầm dừng một pipeline hoặc nhân đôi dữ liệu dễ dàng như một deploy ứng dụng tồi. Xem ./17-cicd-for-data-engineering.md.
• Để dbt sở hữu DAG tầng transformation; để Airflow (hoặc orchestrator của bạn) sở hữu DAG tầng pipeline. Chống lại sự tách biệt này — ví dụ, tái triển khai dependency graph model của dbt thành các task Airflow riêng lẻ cho từng model — thường thêm độ phức tạp vận hành mà không mang lại lợi ích tương xứng, trừ khi bạn cần cụ thể độ chi tiết retry theo từng model mà các công cụ như astronomer-cosmos cung cấp gọn gàng hơn.
• Cảnh báo khi SLA bị miss, không chỉ khi fail. Một pipeline “thành công” nhưng mất thời gian gấp ba lần bình thường là một cảnh báo sớm sẽ trở thành failure nếu không được xử lý.

Tài liệu tham khảo

• Apache Airflow Documentation
• Airflow — DAGs concepts
• Airflow — Scheduler
• dbt Docs — Building models with ref()
• Prefect Documentation
• Dagster Documentation
• Luigi Documentation

0 2 * * * /usr/local/bin/extract.sh && /usr/local/bin/transform.sh && /usr/local/bin/load.sh

extract_orders     extract_customers
      \                   /
       \                 /
        v               v
         transform_and_join
                |
                v
           load_to_warehouse
                |
                v
        refresh_dashboard_mart

┌───────────── minute (0 - 59)
│ ┌───────────── hour (0 - 23)
│ │ ┌───────────── day of month (1 - 31)
│ │ │ ┌───────────── month (1 - 12)
│ │ │ │ ┌───────────── day of week (0 - 6) (Sunday to Saturday)
│ │ │ │ │
* * * * *

from datetime import datetime, timedelta

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
from airflow.sensors.filesystem import FileSensor

default_args = {
    "owner": "data-engineering",
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
    "retry_exponential_backoff": True,
    "max_retry_delay": timedelta(minutes=30),
    "email_on_failure": True,
    "email": ["data-eng-alerts@example.com"],
}

with DAG(
    dag_id="daily_orders_pipeline",
    description="Extract, transform, and load daily order data into the warehouse",
    default_args=default_args,
    schedule="0 2 * * *",       # every day at 02:00
    start_date=datetime(2026, 1, 1),
    catchup=False,              # do not backfill historical runs on deploy
    max_active_runs=1,          # don't let overlapping runs step on each other
    sla_miss_callback=None,     # plug in a Slack/PagerDuty callback here
    tags=["orders", "warehouse"],
) as dag:

    # Wait until the upstream export file has landed before doing anything else.
    wait_for_export = FileSensor(
        task_id="wait_for_export_file",
        filepath="/data/exports/orders_{{ ds_nodash }}.csv",
        poke_interval=60,
        timeout=60 * 60,        # give up after 1 hour
        mode="reschedule",      # free the worker slot between pokes
    )

    def extract_orders(**context):
        """Pull raw orders for this DAG run's data interval and stage them."""
        import pandas as pd

        ds = context["ds"]  # logical date, e.g. "2026-01-01"
        df = pd.read_csv(f"/data/exports/orders_{context['ds_nodash']}.csv")
        df.to_parquet(f"/data/staging/orders_{ds}.parquet", index=False)
        # Push a small summary to XCom for downstream tasks/observability
        context["ti"].xcom_push(key="row_count", value=len(df))

    extract = PythonOperator(
        task_id="extract_orders",
        python_callable=extract_orders,
    )

    def transform_orders(**context):
        """Clean and enrich the staged orders (idempotent: overwrite, not append)."""
        import pandas as pd

        ds = context["ds"]
        df = pd.read_parquet(f"/data/staging/orders_{ds}.parquet")
        df = df.dropna(subset=["order_id", "customer_id"])
        df["order_total"] = df["quantity"] * df["unit_price"]
        df.to_parquet(f"/data/transformed/orders_{ds}.parquet", index=False)

    transform = PythonOperator(
        task_id="transform_orders",
        python_callable=transform_orders,
    )

    # Idempotent load: DELETE + INSERT for this partition, never a blind append,
    # so retries and backfills for the same ds never double-count rows.
    load = BashOperator(
        task_id="load_orders_to_warehouse",
        bash_command=(
            "python /opt/pipelines/load_orders.py "
            "--input /data/transformed/orders_{{ ds }}.parquet "
            "--partition {{ ds }} "
            "--mode upsert"
        ),
    )

    # Trigger the dbt models that depend on the raw orders table.
    run_dbt_models = BashOperator(
        task_id="run_dbt_transform_models",
        bash_command=(
            "cd /opt/dbt_project && "
            "dbt run --select tag:orders --target prod "
            "--vars '{\"run_date\": \"{{ ds }}\"}'"
        ),
    )

    # Structure: wait -> extract -> transform -> load -> dbt models
    wait_for_export >> extract >> transform >> load >> run_dbt_models

-- models/staging/stg_orders.sql
select * from {{ source('raw', 'orders') }}

-- models/marts/fct_orders.sql
select
    o.order_id,
    o.customer_id,
    o.order_total
from {{ ref('stg_orders') }} o   -- this ref() call creates a DAG edge

from prefect import flow, task

@task(retries=3, retry_delay_seconds=30)
def extract_orders(date: str) -> list[dict]:
    ...

@task
def transform(orders: list[dict]) -> list[dict]:
    ...

@task
def load(rows: list[dict]) -> None:
    ...

@flow(name="daily-orders-pipeline")
def daily_orders_pipeline(date: str):
    orders = extract_orders(date)
    clean = transform(orders)
    load(clean)

default_args = {
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
    "retry_exponential_backoff": True,   # 5m, 10m, 20m, ... instead of flat 5m
    "max_retry_delay": timedelta(minutes=30),
}