Data Engineering 3 — Pipeline & Orchestration

Trong hai bài trước chúng ta đã nói về vai trò của Data Engineer và về kho dữ liệu (data warehouse). Bài này đi vào trái tim của nghề: data pipeline — những đường ống vận chuyển dữ liệu từ nơi nó sinh ra tới nơi nó được dùng. Chúng ta sẽ đi từ khái niệm cơ bản nhất tới cách điều phối (orchestration) một hệ thống pipeline thực tế bằng Apache Airflow, và những vấn đề "đau đầu" mà mọi kỹ sư dữ liệu đều phải đối mặt: chạy lại pipeline mà không làm hỏng dữ liệu, batch hay streaming, và làm sao biết dữ liệu là "đúng".

Mục tiêu của bài: sau khi đọc, bạn có thể tự thiết kế một pipeline EOD (end-of-day) cho ngân hàng, viết được một DAG Airflow đơn giản, và giải thích được vì sao idempotency lại quan trọng đến vậy.

Data pipeline là gì?

Một data pipeline là một chuỗi các bước xử lý có thứ tự, lấy dữ liệu từ một hoặc nhiều nguồn, biến đổi nó, rồi đưa tới đích. Hãy tưởng tượng một nhà máy nước: nước thô từ sông (nguồn) đi qua các bể lắng, bể lọc, khử trùng (biến đổi), rồi vào bể chứa và đến vòi nhà bạn (đích). Data pipeline cũng vậy, chỉ khác là "nước" ở đây là dữ liệu.

Một pipeline điển hình có ba đặc tính:

• Có thứ tự (ordered): bước sau phụ thuộc kết quả bước trước. Bạn không thể "transform" trước khi "extract".
• Lặp lại được (repeatable): pipeline thường chạy theo lịch — mỗi giờ, mỗi ngày, mỗi cuối tháng.
• Tự động (automated): con người không ngồi bấm nút từng bước; hệ thống tự chạy và tự báo lỗi.

Đừng nhầm pipeline với một script chạy một lần. Pipeline là thứ sống lâu dài, chạy hàng nghìn lần, và phải xử lý được mọi tình huống xấu: nguồn bị chậm, mạng đứt giữa chừng, dữ liệu bẩn, hay phải chạy lại vì hôm qua sai.

ETL vs ELT

Hai chữ viết tắt này là nền tảng của thiết kế pipeline. Cả hai đều gồm ba việc: Extract (trích xuất), Transform (biến đổi), Load (nạp). Khác biệt nằm ở thứ tự của Transform và Load.

ETL — Extract, Transform, Load

Dữ liệu được trích ra, biến đổi ở một vùng trung gian (thường là máy chủ ETL hoặc engine xử lý riêng), rồi mới nạp vào kho đích đã ở dạng sạch. Đây là cách kinh điển từ thời data warehouse truyền thống, khi dung lượng và sức tính toán của warehouse còn đắt đỏ.

ELT — Extract, Load, Transform

Dữ liệu được trích ra, nạp thẳng vào kho ở dạng thô, rồi mới biến đổi ngay bên trong kho bằng sức mạnh tính toán của kho đó (ví dụ BigQuery, Snowflake, Redshift). Đây là cách hiện đại, thắng thế nhờ kho dữ liệu đám mây có sức tính toán gần như vô hạn và rẻ.

Khi nào dùng ETL? Khi bạn bắt buộc phải làm sạch hoặc che (mask) dữ liệu nhạy cảm trước khi nó chạm vào kho (ví dụ quy định không cho lưu số CMND thô), hoặc khi kho đích yếu.

Khi nào dùng ELT? Hầu hết các trường hợp hiện đại. Giữ raw trong kho cho phép bạn "transform lại" khi phát hiện logic cũ sai mà không phải đi xin lại dữ liệu từ hệ nguồn — một lợi thế cực lớn về truy vết và sửa lỗi.

Ba bước cốt lõi

Extract — Trích xuất

Lấy dữ liệu ra khỏi nguồn: database giao dịch (qua JDBC/CDC), file CSV/Parquet, API, message queue (Kafka), hay log. Vấn đề chính là không làm sập hệ nguồn — bạn không được chạy một câu SELECT * khổng lồ vào core banking giờ cao điểm. Thường ta trích vào giờ thấp điểm hoặc qua bản sao read-replica.

Transform — Biến đổi

Làm sạch (chuẩn hóa định dạng ngày, bỏ trùng), chuẩn hóa (mapping mã chi nhánh), tính toán (tổng dư nợ theo khách), join nhiều nguồn, và áp dụng business rule. Đây là nơi chứa logic nghiệp vụ nặng nhất và cũng dễ sai nhất.

Load — Nạp

Đưa dữ liệu vào đích: warehouse, data mart, hay một bảng phục vụ báo cáo. Hai chiến lược chính: full load (xóa và nạp lại toàn bộ) và incremental load (chỉ nạp phần mới/thay đổi). Ta sẽ bàn kỹ phần incremental ở dưới.

Orchestration là gì và vì sao cần?

Khi bạn có 3 pipeline thì một cron job có thể đủ. Nhưng khi có 300 pipeline với hàng nghìn task phụ thuộc lẫn nhau, cron không cứu nổi bạn. Orchestration là việc điều phối: quyết định task nào chạy khi nào, theo thứ tự nào, xử lý lỗi ra sao, và cho bạn thấy toàn cảnh.

Bốn lý do then chốt cần orchestration:

• Phụ thuộc (dependency): task "tính lãi" chỉ được chạy sau khi task "nạp số dư" xong. Orchestrator biểu diễn quan hệ này rõ ràng và đảm bảo đúng thứ tự.
• Lịch (scheduling): chạy hằng ngày lúc 1 giờ sáng, hằng giờ, hay cuối tháng. Orchestrator quản lý lịch tập trung.
• Retry: task lỗi do mạng chập chờn? Tự thử lại 3 lần, mỗi lần cách 5 phút, trước khi báo người.
• Giám sát (monitoring): thấy ngay task nào fail, chạy bao lâu, log ở đâu, và cảnh báo qua email/Slack khi có sự cố.

Apache Airflow

Airflow là công cụ orchestration phổ biến nhất, viết bằng Python, mã nguồn mở (ban đầu từ Airbnb). Pipeline trong Airflow được biểu diễn bằng DAG.

DAG — Directed Acyclic Graph

DAG (đồ thị có hướng, không chu trình) là cách Airflow mô tả một pipeline: các node là task, các mũi tên là quan hệ phụ thuộc. "Có hướng" nghĩa là mũi tên chỉ một chiều (A chạy trước B). "Không chu trình" nghĩa là không được vòng lại (A → B → A là cấm), nếu không pipeline sẽ chạy mãi không dừng.

Task và Operator

Một task là một đơn vị công việc. Mỗi task được tạo từ một operator — một template định nghĩa kiểu việc:

• PythonOperator — chạy một hàm Python.
• BashOperator — chạy lệnh shell.
• SQLExecuteQueryOperator — chạy câu SQL trên một database.
• Các operator chuyên dụng cho GCP, AWS, Spark, v.v.

Scheduler

Scheduler là bộ não của Airflow: nó đọc tất cả DAG, biết DAG nào tới giờ chạy, kiểm tra task nào đã đủ điều kiện phụ thuộc, rồi đẩy task vào hàng đợi để worker thực thi.

Dependency

Bạn khai báo phụ thuộc bằng toán tử >> (downstream): extract >> transform >> load nghĩa là extract xong mới transform, transform xong mới load.

Backfill

Backfill là chạy lại pipeline cho các mốc thời gian trong quá khứ. Ví dụ: bạn vừa viết một DAG mới và muốn nó tính lại dữ liệu cho cả tháng trước. Airflow cho phép backfill từng "ngày logic" (logical date) một cách có hệ thống. Đây là lý do idempotency trở nên sống còn — chạy backfill nghĩa là chạy lại, nhiều lần.

Ví dụ DAG Airflow

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator

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

def extract(**context):
    # ds = ngày logic, ví dụ "2026-06-29"
    ds = context["ds"]
    print(f"Trich du lieu giao dich ngay {ds} tu core banking -> staging")

def transform(**context):
    ds = context["ds"]
    print(f"Lam sach + tinh tong du no theo khach cho ngay {ds}")

def load(**context):
    ds = context["ds"]
    # Idempotent: xoa partition cua ngay nay truoc khi nap lai
    print(f"DELETE partition WHERE date = {ds}; INSERT du lieu da transform")

with DAG(
    dag_id="eod_loan_pipeline",
    start_date=datetime(2026, 1, 1),
    schedule="0 1 * * *",        # 1h sang moi ngay
    catchup=False,
    default_args=default_args,
    tags=["eod", "banking"],
) as dag:

    t_extract = PythonOperator(task_id="extract", python_callable=extract)
    t_transform = PythonOperator(task_id="transform", python_callable=transform)
    t_load = PythonOperator(task_id="load", python_callable=load)

    t_extract >> t_transform >> t_load

Lưu ý ba điểm: retries=3 để tự thử lại khi lỗi tạm thời; schedule="0 1 * * *" là cron chạy 1h sáng; và task load thiết kế idempotent bằng cách xóa partition của ngày đó trước khi nạp lại.

Idempotency và exactly-once

Idempotency nghĩa là: chạy một thao tác một lần hay nhiều lần đều cho cùng một kết quả. Đây là tính chất quan trọng nhất của một pipeline tốt.

Vì sao? Vì pipeline sẽ chạy lại. Mạng đứt giữa chừng, scheduler restart, bạn backfill, ai đó bấm "retry". Nếu task load của bạn là INSERT thuần, chạy 2 lần sẽ nhân đôi dữ liệu. Nếu nó tính "cộng dồn doanh thu", chạy lại sẽ cộng dồn 2 lần — sai be bét.

Các kỹ thuật làm pipeline idempotent:

• Overwrite theo partition: thay vì INSERT, hãy DELETE WHERE date = ds rồi INSERT. Chạy lại bao nhiêu lần, partition ngày đó vẫn đúng một bản.
• UPSERT / MERGE: dùng MERGE (insert nếu chưa có, update nếu đã có) theo khóa chính.
• Khóa tự nhiên + dedup: nạp thô rồi loại trùng theo business key.

Exactly-once là mục tiêu lý tưởng: mỗi bản ghi được xử lý đúng một lần, không thiếu không thừa. Trong thực tế, nhiều hệ thống chỉ đảm bảo at-least-once (ít nhất một lần, có thể trùng). Mẹo phổ biến là: chấp nhận at-least-once ở tầng truyền tải, rồi dùng idempotency ở tầng xử lý để biến nó thành exactly-once về mặt hiệu quả. Nói cách khác, idempotency là con đường thực tế để đạt exactly-once.

Batch vs Streaming

Có hai mô hình xử lý dữ liệu cơ bản, khác nhau ở chỗ dữ liệu được gom thành "lô" hay chảy "liên tục".

Quy tắc ngón tay cái: nếu nghiệp vụ chấp nhận độ trễ vài phút tới vài giờ (báo cáo, đối soát), dùng batch — đơn giản và rẻ. Nếu cần phản ứng tức thì (chặn giao dịch nghi ngờ gian lận trong vài trăm mili-giây), dùng streaming. Đa số ngân hàng dùng batch cho EOD và streaming cho fraud/realtime alert.

Incremental load và CDC

Mỗi ngày nạp lại toàn bộ bảng giao dịch 5 năm là cực kỳ lãng phí. Incremental load chỉ xử lý phần mới hoặc thay đổi kể từ lần chạy trước.

Cách đơn giản nhất là dùng cột mốc thời gian: "lấy mọi bản ghi có updated_at > lần chạy trước". Nhưng cách này bỏ sót việc xóa (DELETE không cập nhật updated_at) và phụ thuộc vào việc hệ nguồn cập nhật cột này đúng.

CDC (Change Data Capture) là kỹ thuật cao cấp hơn: thay vì truy vấn bảng, ta đọc trực tiếp transaction log của database nguồn (binlog của MySQL, WAL của PostgreSQL). Mọi thay đổi — insert, update, delete — đều được bắt lại dưới dạng sự kiện. Công cụ như Debezium đẩy các sự kiện này vào Kafka, từ đó pipeline có thể đồng bộ gần như realtime mà không gây tải truy vấn lên hệ nguồn. CDC là cầu nối tự nhiên giữa thế giới batch và streaming.

Data quality check trong pipeline

Một pipeline chạy thành công về mặt kỹ thuật chưa có nghĩa dữ liệu đúng. Bạn cần chèn các bước kiểm tra chất lượng:

• Kiểm tra số lượng (volume): hôm nay có 1.2 triệu giao dịch, hôm qua 1.1 triệu — hợp lý. Nếu hôm nay chỉ có 50, chắc chắn extract bị lỗi.
• Tính toàn vẹn (not null, unique): cột account_id không được rỗng; khóa chính phải duy nhất.
• Giá trị hợp lệ (range/domain): số dư không âm; mã chi nhánh nằm trong danh sách hợp lệ.
• Đối soát (reconciliation): tổng số dư trong warehouse phải khớp tổng từ core banking trong sai số cho phép.

Triết lý quan trọng: fail fast. Nếu kiểm tra fail, hãy dừng pipeline và cảnh báo, đừng để dữ liệu sai chảy xuống tới dashboard giám đốc. Công cụ như Great Expectations hay dbt tests giúp khai báo các check này một cách có hệ thống.

Ví dụ luồng EOD ngân hàng

Hãy ghép tất cả lại thành một luồng EOD (cuối ngày) điển hình: trích từ core banking, đưa vào staging, transform, rồi nạp vào warehouse — với data quality check chặn ở giữa.

Diễn giải luồng:

1. Extract: vào giờ thấp điểm (ví dụ 1h sáng), trích giao dịch trong ngày từ core banking qua read-replica để không ảnh hưởng hệ thống giao dịch.
2. Staging: nạp thô vào vùng staging — đây là tinh thần ELT, giữ raw để truy vết.
3. DQ check: kiểm tra số lượng bản ghi và các cột bắt buộc không null. Fail thì dừng và báo.
4. Transform: làm sạch, chuẩn hóa, tính tổng dư nợ theo khách hàng, theo chi nhánh.
5. Reconciliation: đối soát tổng số dư đã tính với con số từ core. Lệch ngoài ngưỡng thì dừng.
6. Load: nạp idempotent (overwrite partition theo ngày) vào warehouse.
7. Data mart: phục vụ báo cáo EOD cho ban lãnh đạo.

Toàn bộ luồng này được điều phối bởi một DAG Airflow, chạy theo lịch, tự retry các bước extract khi mạng chập chờn, và cảnh báo qua email nếu DQ check fail.

Tóm tắt

Data pipeline là chuỗi bước có thứ tự, lặp lại và tự động để đưa dữ liệu từ nguồn tới đích. ETL transform trước khi nạp (phù hợp khi kho yếu hoặc cần che dữ liệu nhạy cảm); ELT nạp thô trước rồi transform trong kho (phổ biến nhất hiện nay nhờ kho đám mây mạnh và giữ được raw để truy vết). Orchestration giải quyết phụ thuộc, lịch, retry và giám sát; Airflow biểu diễn pipeline bằng DAG gồm các task/operator, do scheduler điều phối, hỗ trợ backfill. Idempotency — chạy lại cho cùng kết quả — là tính chất sống còn vì pipeline luôn bị chạy lại, và là con đường thực tế để đạt exactly-once. Batch đơn giản, rẻ, độ trễ cao; streaming độ trễ thấp nhưng phức tạp. Incremental load và CDC giúp chỉ xử lý phần thay đổi, tiết kiệm tài nguyên. Cuối cùng, data quality check với triết lý fail-fast bảo vệ dữ liệu khỏi chảy sai xuống hạ nguồn.

Tự kiểm tra

1. Khác biệt cốt lõi giữa ETL và ELT là gì, và vì sao ELT thắng thế trong kỷ nguyên kho dữ liệu đám mây?
2. Một DAG trong Airflow có thể chứa chu trình (A → B → A) không? Vì sao?
3. Giải thích vì sao idempotency lại quan trọng đến vậy, và nêu hai kỹ thuật để làm task load trở nên idempotent.
4. Khi nào bạn chọn batch, khi nào chọn streaming? Cho một ví dụ ngân hàng cho mỗi loại.
5. CDC khác gì so với incremental load bằng cột updated_at, và CDC giải quyết được vấn đề gì mà cách kia bỏ sót?
6. Trong luồng EOD, vì sao bước reconciliation nên được đặt trước bước load vào warehouse?

Đọc tiếp

Data Engineering 4 — Apache Spark