[Tech Blog #12 | Data Engineering] Tự Động Hóa Pipeline ETL Đầu Tiên Bằng Python: Trích Xuất Dữ Liệu REST API & Nạp Vào CSDL

---

1. Đặt vấn đề: Nỗi ám ảnh của việc "Tải file Excel thủ công mỗi sáng"

Hãy tưởng tượng công việc hàng ngày của một nhân viên vận hành:

• Mỗi sáng đúng 8h00: Đăng nhập vào cổng thanh toán đối tác hoặc hệ thống CRM.
• Bấm nút "Xuất báo cáo CSV" của ngày hôm trước.
• Mở file Excel, xóa vài cột rác, sửa lại định dạng ngày tháng.
• Mở hệ thống cơ sở dữ liệu nội bộ và bấm Import thủ công.

Quy trình này không chỉ gây nhàm chán mà còn tiềm ẩn vô số rủi ro: người thực hiện có thể bị ốm hoặc quên, file xuất ra bị đổi định dạng gây lỗi, hoặc nếu dữ liệu tăng lên hàng triệu dòng thì Excel sẽ treo cứng ngắc.

Nhiệm vụ cốt lõi của một Data Engineer là giải phóng con người khỏi những thao tác thủ công đó bằng cách xây dựng các Đường ống dữ liệu tự động (Data Pipelines) vận hành theo nguyên lý ETL (Extract - Transform - Load).

---

2. Giải mã chu trình ETL thực chiến

[REST API Nguồn] 
       │
       ▼ (1. EXTRACT: Gọi HTTP GET, bắt lỗi mạng, xử lý phân trang)
[Dữ liệu JSON thô]
       │
       ▼ (2. TRANSFORM: Làm phẳng JSON, chuẩn hóa kiểu dữ liệu, gán nhãn thời gian)
[Dữ liệu sạch chuẩn hóa]
       │
       ▼ (3. LOAD: Nạp vào Database với kỹ thuật Upsert / Idempotency)
[Cơ sở dữ liệu (PostgreSQL / SQLite / Data Warehouse)]

2.1. Extract (Trích xuất dữ liệu)

• Kết nối tới các nguồn dữ liệu bên ngoài (phổ biến nhất là qua REST API trả về định dạng JSON).
• Thách thức thực tế: Xử lý giới hạn tần suất gọi (Rate Limiting), cơ chế thử lại khi mất mạng (Retry / Exponential Backoff), và duyệt qua nhiều trang kết quả (Pagination).

2.2. Transform (Làm sạch & Biến đổi)

• Bóc tách cấu trúc lồng nhau (Flatten nested JSON).
• Ép kiểu dữ liệu chuẩn (ví dụ: chuyển chuỗi "2026-09-23T08:00:00Z" thành kiểu Timestamp, chuyển chuỗi tiền tệ thành số thực Float).
• Bổ sung các siêu dữ liệu quản trị (Audit Columns): Ví dụ thêm cột ingested_at để biết bản ghi này được nạp vào hệ thống lúc mấy giờ.

2.3. Load (Nạp dữ liệu vào Database)

• Ghi dữ liệu đã làm sạch vào cơ sở dữ liệu đích (PostgreSQL, MySQL, ClickHouse).
• Khái niệm sống còn - Idempotency (Tính lũy thừa):
  Một pipeline chuẩn phải đảm bảo: Dù bạn có chạy script 1 lần hay lỡ tay chạy 10 lần liên tiếp thì dữ liệu trong database vẫn hoàn toàn chính xác, không bị nhân bản (duplicate) gấp 10 lần.
  Để làm được điều này, chúng ta sử dụng kỹ thuật Upsert (INSERT ... ON CONFLICT DO UPDATE).

---

3. Bảng so sánh: ETL vs. ELT

Trong thế giới kỹ thuật dữ liệu hiện đại, bạn sẽ thường xuyên nghe thấy hai khái niệm ETL và ELT:

Tiêu chí	ETL (Extract - Transform - Load)	ELT (Extract - Load - Transform)
Quy trình	Biến đổi dữ liệu trên máy chủ trung gian (RAM/Python) trước khi nạp vào Database.	Nạp toàn bộ dữ liệu thô vào Data Warehouse trước, sau đó dùng SQL (dbt) để biến đổi.
Công nghệ tiêu biểu	Python scripts, Apache Spark, Pentaho.	Snowflake, BigQuery, ClickHouse kết hợp dbt.
Khả năng lưu trữ	Chỉ lưu dữ liệu đã qua xử lý, tiết kiệm bộ nhớ đích.	Lưu toàn bộ lịch sử thô (Data Lake), cho phép phân tích lại dữ liệu cũ bất kỳ lúc nào.
Khi nào nên dùng?	Hệ thống nhỏ đến vừa, cần ẩn danh hóa dữ liệu nhạy cảm (PII/Masking) trước khi lưu.	Hệ thống Big Data hiện đại, tận dụng sức mạnh tính toán song song cực mạnh của Cloud DWH.

---

4. Thực hành thực chiến: Xây dựng Pipeline ETL hoàn chỉnh bằng Python

Dưới đây là một script Python độc lập, sử dụng thư viện requests để lấy dữ liệu từ một REST API công khai (JSONPlaceholder) và nạp dữ liệu sạch vào CSDL SQLite/PostgreSQL với cơ chế Upsert chống trùng lặp:

import sqlite3
import datetime
import requests

# ==============================================================================
# BƯỚC 1: EXTRACT (Trích xuất dữ liệu từ REST API)
# ==============================================================================
def extract_data_from_api(api_url: str) -> list:
    """Gọi REST API với timeout và kiểm tra HTTP status code."""
    print(f"[{datetime.datetime.now()}] Đang gọi API: {api_url}...")
    try:
        response = requests.get(api_url, timeout=10)
        # Ném ra lỗi nếu HTTP Status Code là 4xx hoặc 5xx
        response.raise_for_status()
        data = response.json()
        print(f"Trích xuất thành công {len(data)} bản ghi thô từ nguồn.")
        return data
    except requests.exceptions.RequestException as e:
        print(f"LỖI EXTRACT: Không thể kết nối tới API. Chi tiết: {e}")
        return []

# ==============================================================================
# BƯỚC 2: TRANSFORM (Làm sạch, phẳng hóa JSON & Chuẩn hóa kiểu dữ liệu)
# ==============================================================================
def transform_user_data(raw_users: list) -> list:
    """Làm phẳng cấu trúc JSON lồng nhau và chuẩn hóa các trường dữ liệu."""
    print(f"[{datetime.datetime.now()}] Bắt đầu quá trình biến đổi (Transform)...")
    transformed_records = []
    current_timestamp = datetime.datetime.utcnow().isoformat()
    
    for item in raw_users:
        # Bóc tách cấu trúc lồng nhau (Flatten nested dictionary)
        user_id = int(item.get("id"))
        name = str(item.get("name", "")).strip()
        email = str(item.get("email", "")).lower().strip()
        
        # Trích xuất địa chỉ lồng bên trong dict 'address'
        address_info = item.get("address", {})
        city = str(address_info.get("city", "Unknown")).strip()
        
        # Trích xuất tên công ty lồng bên trong dict 'company'
        company_info = item.get("company", {})
        company_name = str(company_info.get("name", "Individual")).strip()
        
        # Đưa vào tuple chuẩn sẵn sàng nạp vào Database
        record = (user_id, name, email, city, company_name, current_timestamp)
        transformed_records.append(record)
        
    print(f"Biến đổi hoàn tất {len(transformed_records)} bản ghi hợp lệ.")
    return transformed_records

# ==============================================================================
# BƯỚC 3: LOAD (Nạp vào Cơ sở dữ liệu với cơ chế UPSERT chống trùng lặp)
# ==============================================================================
def load_to_database(db_path: str, records: list):
    """Khởi tạo bảng và nạp dữ liệu với cơ chế IDEMPOTENT (Chống trùng lặp)."""
    print(f"[{datetime.datetime.now()}] Bắt đầu nạp dữ liệu vào Database: {db_path}...")
    
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    
    # 1. Tạo bảng đích nếu chưa tồn tại
    create_table_query = """
    CREATE TABLE IF NOT EXISTS dim_users (
        user_id INTEGER PRIMARY KEY,
        full_name TEXT NOT NULL,
        email TEXT NOT NULL,
        city TEXT,
        company_name TEXT,
        updated_at TIMESTAMP
    );
    """
    cursor.execute(create_table_query)
    
    # 2. Câu lệnh UPSERT: Nếu user_id đã tồn tại thì UPDATE, chưa có thì INSERT
    # Cú pháp SQLite >= 3.24 & PostgreSQL tương đồng
    upsert_query = """
    INSERT INTO dim_users (user_id, full_name, email, city, company_name, updated_at)
    VALUES (?, ?, ?, ?, ?, ?)
    ON CONFLICT(user_id) DO UPDATE SET
        full_name = excluded.full_name,
        email = excluded.email,
        city = excluded.city,
        company_name = excluded.company_name,
        updated_at = excluded.updated_at;
    """
    
    # Thực thi nạp hàng loạt (Batch Execution) để tối ưu hiệu năng
    cursor.executemany(upsert_query, records)
    conn.commit()
    
    # Kiểm tra số dòng thực tế trong bảng sau khi nạp
    cursor.execute("SELECT COUNT(*) FROM dim_users;")
    total_rows = cursor.fetchone()[0]
    print(f"Nạp hoàn tất! Tổng số người dùng hiện có trong Database: {total_rows}")
    
    conn.close()

# ==============================================================================
# HÀM ĐIỀU PHỐI CHÍNH (PIPELINE RUNNER)
# ==============================================================================
def run_etl_pipeline():
    TARGET_API = "[https://jsonplaceholder.typicode.com/users](https://jsonplaceholder.typicode.com/users)"
    DATABASE_FILE = "analytics_warehouse.db"
    
    # Thực thi tuần tự 3 bước
    raw_data = extract_data_from_api(TARGET_API)
    if not raw_data:
        print("Pipeline dừng lại do không trích xuất được dữ liệu.")
        return
        
    clean_data = transform_user_data(raw_data)
    load_to_database(DATABASE_FILE, clean_data)
    print("--- PIPELINE ETL HOÀN TẤT THÀNH CÔNG RỰC RỠ ---")

if __name__ == "__main__":
    # Chạy thử lần 1
    print(">>> CHẠY PIPELINE LẦN 1:")
    run_etl_pipeline()
    
    # Chạy thử lần 2 ngay sau đó để chứng minh tính Idempotent (Không bị phình to dữ liệu)
    print("\n>>> CHẠY PIPELINE LẦN 2 (KIỂM TRA TÍNH LŨY THỪA / IDEMPOTENCY):")
    run_etl_pipeline()

---

5. Câu hỏi phỏng vấn thực tế (Interview Q&A)

Câu 1 (Mức độ: Entry / Junior)

Câu hỏi: "Tính chất Idempotency (Tính lũy thừa) trong Data Pipeline là gì? Làm thế nào bạn đảm bảo được tính chất này khi viết code nạp dữ liệu?"

Gợi ý trả lời:
*"Idempotency là tính chất đảm bảo rằng: khi một đường ống dữ liệu (pipeline) được thực thi nhiều lần với cùng một tập dữ liệu đầu vào, kết quả cuối cùng trong cơ sở dữ liệu đích luôn giống hệt như khi chỉ chạy một lần duy nhất, không gây ra lỗi trùng lặp dữ liệu hay sai lệch số liệu.
Để đảm bảo tính Idempotent trong thực tế, em thường áp dụng:

1. Cơ chế Upsert (ON CONFLICT DO UPDATE / MERGE): Dựa vào khóa chính (Primary Key) hoặc Khóa tự nhiên (Natural Key) của bản ghi để cập nhật thay vì chèn dòng mới.
2. Chiến lược Partition Overwrite: Với các bảng dữ liệu theo ngày, trước khi nạp dữ liệu ngày hôm đó, ta xóa sạch (hoặc ghi đè) đúng partition của ngày đó thay vì dùng INSERT INTO thông thường."*

Câu 2 (Mức độ: Junior / Mid-level)

Câu hỏi: "Khi cào dữ liệu từ API bên thứ ba, nếu hệ thống đối tác trả về mã lỗi HTTP 429 (Too Many Requests) hoặc mạng bị chập chờn, bạn sẽ xử lý như thế nào để pipeline không bị sập?"

Gợi ý trả lời:
*"Để xử lý tình huống này, em sẽ triển khai chiến lược Retry with Exponential Backoff and Jitter:

1. Khi gặp mã lỗi 429 hoặc 503 Service Unavailable, script không nên thoát ngay mà sẽ tạm dừng (sleep) một khoảng thời gian ngắn (ví dụ: 1s), sau đó thử gọi lại lần 2.
2. Nếu vẫn thất bại, thời gian chờ sẽ tăng theo cấp số nhân (2s, 4s, 8s, 16s...) kết hợp với một khoảng thời gian ngẫu nhiên nhỏ (Jitter) để tránh việc tất cả các worker cùng đồng loạt bắn request vào server đối tác cùng lúc (hiện tượng Thundering Herd).
3. Thiết lập số lần thử tối đa (ví dụ: max_retries = 5). Nếu vượt quá số lần này, hệ thống sẽ log cảnh báo chi tiết và bắn thông báo qua Slack/Telegram hoặc gửi mail cho đội ngũ trực để can thiệp thủ công."*

---

6. Key Takeaways (Ghi nhớ nhanh)

1. ETL giải phóng con người: Tự động hóa các tác vụ trích xuất, biến đổi và nạp dữ liệu định kỳ với độ tin cậy tuyệt đối.
2. Luôn thiết kế Pipeline có tính Idempotent: Hãy dùng Upsert hoặc ghi đè theo phân vùng (Partition Overwrite) để khi chạy lại script không bao giờ bị nhân đôi dữ liệu.
3. Bảo vệ hệ thống bằng Audit Columns: Luôn thêm cột updated_at hoặc ingested_at vào bảng đích để dễ dàng đối soát và gỡ lỗi khi có sự cố.
4. Xử lý ngoại lệ là bắt buộc: Luôn bọc các lệnh gọi API bằng try...except, thiết lập timeout rõ ràng và áp dụng cơ chế Retry khi gặp sự cố mạng.