[DE Blog #64] Tự Động Hóa Thu Thập Dữ Liệu Bằng Python (Data Ingestion 101): Gọi REST API, Xử Lý JSON Lồng Nhau & Viết Script ETL Nạp Vào CSDL Chuẩn Idempotent

1. Bối cảnh thực tế (Context & Problem Statement)

Hãy tưởng tượng bạn mới vào làm việc tại một công ty thương mại điện tử hoặc tài chính:

Mỗi sáng, sếp yêu cầu bạn cập nhật: Tỷ giá hối đoái ngoại tệ từ ngân hàng, Giá vàng, hoặc Dữ liệu thời tiết của các kho hàng trên cả nước.
Cách làm thủ công của người nghiệp dư: Hàng ngày hẹn giờ thức dậy lúc 7:00 sáng, vào trang web, bấm tải file Excel về máy, copy-paste vào trang tính rồi nạp lên báo cáo.

Những rủi ro tai hại của việc làm thủ công:

Lãng phí thời gian và dễ sai sót do con người: Quên tải dữ liệu một ngày, copy nhầm cột, hoặc mạng chập chờn tải thiếu file.
Không thể mở rộng (Cannot Scale): Nếu công ty cần theo dõi $500$ chi nhánh hoặc lấy dữ liệu tỷ giá cập nhật mỗi $5\text{ phút}$ , con người hoàn toàn bất lực.

Là một kỹ sư dữ liệu (Data Engineer), nhiệm vụ đầu tay của bạn là viết một Script Python tự động hóa hoàn toàn chu trình ETL (Extract - Transform - Load):

Tự động gọi API lấy dữ liệu nhị phân/JSON về máy (Extract).
Làm sạch, bóc tách cấu trúc lồng nhau thành bảng phẳng (Transform).
Tự động nạp dữ liệu vào cơ sở dữ liệu (Load) sao cho dù script có chạy lặp lại 10 lần thì dữ liệu cũng không bao giờ bị nhân bản trùng lặp!

2. Các Khái Niệm & Cơ Chế Cốt Lõi

2.1. Bản Chất Chu Trình ETL Mini Bằng Python

[BƯỚC 1: EXTRACT (Trích xuất)]       [BƯỚC 2: TRANSFORM (Biến đổi)]      [BƯỚC 3: LOAD (Nạp dữ liệu)]
+---------------------------+        +---------------------------+       +-------------------------+
| RESTful API (Web Service) |        | Thư viện Pandas           |       | Cơ Sở Dữ Liệu (Database)|
| Gọi requests.get()        | ====>  | Làm phẳng Nested JSON     | ====> | SQLite / PostgreSQL     |
| Trả về chuỗi JSON thô     |        | pd.json_normalize()       |       | Kỹ thuật UPSERT         |
+---------------------------+        +---------------------------+       +-------------------------+

2.2. Giao Tiếp Với RESTful API Bằng Thư Viện `requests`

API (Application Programming Interface) là "người phục vụ bàn" giúp chương trình Python của bạn giao tiếp và lấy dữ liệu từ máy chủ khác thông qua giao thức HTTP:

Phương thức GET: Yêu cầu máy chủ trả về dữ liệu (phổ biến nhất khi làm Data Ingestion).
URL Parameters (params): Gửi các tham số lọc trên thanh địa chỉ (ví dụ: ?currency=USD&date=2026-09-15).
Headers & Authentication: Thẻ căn cước định danh của ứng dụng. Hầu hết các API bảo mật đều yêu cầu truyền mã khóa truy cập (API Key / Bearer Token) vào trong Header để xác thực danh tính.

Các Mã Trạng Thái HTTP (Status Codes) Cần Nằm Lòng:

200 OK: Thành công rực rỡ, dữ liệu đã sẵn sàng.
400 Bad Request: Client gửi tham số bị sai định dạng.
401 Unauthorized / 403 Forbidden: Sai API Key hoặc không có quyền truy cập.
404 Not Found: Đường dẫn URL không tồn tại.
429 Too Many Requests (CỰC KỲ QUAN TRỌNG): Bạn vừa bị máy chủ chặn vì gọi API quá nhiều lần trong 1 giây (Vượt quá Rate Limit)!
500 / 502 / 503 Server Error: Máy chủ của đối tác bị sập hoặc quá tải.

2.3. Bóc Tách Dữ Liệu JSON Lồng Nhau (Nested JSON Flattening)

Dữ liệu từ API không bao giờ trả về dạng bảng thẳng hàng cột lối như SQL. Nó thường trả về ở dạng các cấu trúc lồng nhau phức tạp:

{
  "user_id": 101,
  "name": "Nguyen Van An",
  "contact": {
    "email": "[email protected]",
    "phone": "0903123456"
  },
  "orders": [
    {"order_id": "DH01", "total": 500000},
    {"order_id": "DH02", "total": 250000}
  ]
}

Cứu cánh từ hàm pd.json_normalize() của Pandas:
- Tự động biến đổi các từ điển con (Nested Dict) thành các cột riêng biệt có dấu chấm: contact.email, contact.phone.
- Cho phép chỉ định danh sách mảng (Lists) để tách thành nhiều dòng tương ứng một cách tự động!

2.4. Cơ Chế Tự Động Thử Lại (Retry With Exponential Backoff)

Khi gọi API qua mạng Internet, hiện tượng chập chờn mạng hoặc nghẽn tạm thời là chuyện thường ngày:

Sai lầm của người mới: Gặp lỗi mạng là để script văng lỗi và dừng chương trình (Crash).
Giải pháp chuẩn kỹ sư: Exponential Backoff (Thử lại với thời gian chờ tăng theo cấp số nhân):
- Lần lỗi 1: Ngủ $1\text{ giây}$ rồi thử lại.
- Lần lỗi 2: Ngủ $2\text{ giây}$ rồi thử lại.
- Lần lỗi 3: Ngủ $4\text{ giây}$ rồi thử lại.
- Lần lỗi 4: Ngủ $8\text{ giây}$ ... Nếu thử quá 5 lần vẫn lỗi mới chính thức ghi log cảnh báo.
  $\rightarrow$ Tránh làm sập thêm máy chủ của đối tác và tăng độ tin cậy của Pipeline lên $99.9\%$ .

2.5. Nguyên Tắc Vàng Idempotency & Kỹ Thuật UPSERT

Định nghĩa Idempotency (Tính lũy đẳng): Một tác vụ được gọi là Idempotent nếu việc thực thi nó 1 lần hay 100 lần liên tiếp đều cho ra kết quả cuối cùng hoàn toàn giống nhau mà không sinh ra dữ liệu rác.
Cạm bẫy của lệnh INSERT thông thường: Nếu script chạy lúc 8:00 bị lỗi mạng ở giữa chừng, khi bạn chạy lại script lúc 8:15, các dòng dữ liệu cũ sẽ bị nạp đúp lần thứ 2 (Duplicate Records).
Kỹ thuật UPSERT (INSERT ... ON CONFLICT DO UPDATE):
- Nếu bản ghi đó chưa tồn tại (chưa có Primary Key này): Thực hiện INSERT mới.
- Nếu bản ghi đó đã tồn tại: Tự động chuyển thành lệnh UPDATE cập nhật các giá trị mới nhất.
- $\rightarrow$ Pipeline hoàn toàn an toàn khi chạy lại nhiều lần (Safe to re-run)!

3. Các Bảng Markdown So Sánh Chi Tiết

📊 Bảng 1: So Sánh Thu Thập Dữ Liệu Qua REST API vs. Web Scraping (Cào HTML)

Tiêu chí	Thu thập qua RESTful API	Cào dữ liệu Web (Web Scraping / HTML Parsing)
Độ ổn định kỹ thuật	Rất cao (Được nhà cung cấp cam kết cấu trúc)	Rất kém (Website đổi giao diện/class CSS là script chết ngay)
Định dạng dữ liệu trả về	Dữ liệu có cấu trúc chuẩn (JSON hoặc XML)	Dữ liệu thô hỗn tạp (HTML, DOM tags)
Tốc độ thực thi	Siêu nhanh (Chỉ truyền dữ liệu thô)	Chậm hơn (Phải tải toàn bộ mã HTML, ảnh, CSS của trang)
Rủi ro pháp lý & Chặn IP	Thấp (Được cấp phép chính thức qua API Key)	Rất cao (Dễ bị chặn bởi Cloudflare, Captcha, vi phạm ToS)
Ưu tiên sử dụng	Luôn luôn là lựa chọn ưu tiên số 1	Chỉ dùng khi đối tác hoàn toàn không cung cấp API

📊 Bảng 2: Các Mã Trạng Thái HTTP (Status Codes) Thường Gặp & Cách Xử Lý

Mã HTTP Status	Tên trạng thái	Ý nghĩa bản chất	Hành động chuẩn trong Script Python
`200 OK`	Thành công	Dữ liệu trả về hoàn hảo	Parse JSON và chuyển sang bước Transform
`401 / 403`	Unauthorized / Forbidden	API Key hết hạn hoặc sai quyền	Báo động dừng pipeline ngay, kiểm tra file `.env`
`404`	Not Found	Sai URL hoặc ID đối tượng không tồn tại	Bỏ qua bản ghi đó và ghi nhật ký cảnh báo
`429`	Too Many Requests	Bị máy chủ tạm khóa do bắn request quá nhanh	Bắt buộc dùng `time.sleep()` chờ vài giây trước khi gọi tiếp
`500 / 503`	Internal Server Error	Máy chủ của đối tác bị lỗi nội bộ hoặc bảo trì	Kích hoạt cơ chế Retry with Exponential Backoff

📊 Bảng 3: So Sánh 3 Chiến Lược Nạp Dữ Liệu Vào Cơ Sở Dữ Liệu (Load Strategies)

Chiến lược nạp	Cơ chế thực hiện	Điểm mạnh	Điểm yếu	Trường hợp sử dụng tối ưu
Full Refresh (Ghi đè toàn bộ)	Xóa sạch bảng cũ (`TRUNCATE`), nạp lại toàn bộ	Rất đơn giản, không sợ lệch số liệu	Cực kỳ tốn tài nguyên khi dữ liệu lớn	Bảng danh mục nhỏ ( $< 10,000$ dòng: Bảng Tỉnh thành)
Append Only (Chỉ thêm mới)	Luôn chạy `INSERT INTO` ở cuối bảng	Tốc độ ghi nhanh nhất	Bị trùng lặp nếu chạy lại kịch bản lỗi	Dữ liệu sự kiện bất biến (Clickstream Logs, IoT Metrics)
Upsert (Merge)	Khớp khóa chính: Có thì Update, chưa có thì Insert	Đảm bảo tính Lũy đẳng (Idempotent), dữ liệu sạch 100%	Tốn chi phí kiểm tra index khóa chính	Dữ liệu giao dịch thay đổi trạng thái (Orders, Users)

4. Thực Hành Thực Chiến: Script Python ETL Hoàn Chỉnh Từng Dòng Lệnh

📌 Kịch bản bài toán:

Viết một script Python tự động lấy dữ liệu thời tiết thực tế từ Open-Meteo API (API công cộng miễn phí), làm phẳng dữ liệu JSON và nạp vào cơ sở dữ liệu SQLite theo chuẩn Upsert (Idempotent):

import time
import requests
import pandas as pd
import sqlite3

# ==========================================
# BƯỚC 1: EXTRACT - GỌI API KÈM RETRY LOGIC
# ==========================================
def extract_weather_data(latitude: float, longitude: float, max_retries: int = 3) -> dict:
    """Gọi API lấy dữ liệu thời tiết kèm cơ chế tự động thử lại (Exponential Backoff)."""
    url = "[https://api.open-meteo.com/v1/forecast](https://api.open-meteo.com/v1/forecast)"
    params = {
        "latitude": latitude,
        "longitude": longitude,
        "current_weather": True,
        "hourly": "temperature_2m,relative_humidity_2m"
    }
    
    for attempt in range(1, max_retries + 1):
        try:
            print(f"-> Đang gọi API (Lần thử {attempt}/{max_retries})...")
            response = requests.get(url, params=params, timeout=10) # Timeout 10s tránh treo script
            
            # Nếu gặp mã lỗi 429 (Rate limit) hoặc 5xx (Lỗi server):
            if response.status_code in [429, 500, 502, 503, 504]:
                wait_time = 2 ** attempt # Thời gian chờ tăng theo cấp số nhân: 2s, 4s, 8s
                print(f"Gặp mã lỗi HTTP {response.status_code}. Tạm nghỉ {wait_time}s rồi thử lại...")
                time.sleep(wait_time)
                continue
                
            # Kiểm tra các lỗi HTTP 4xx/5xx khác:
            response.raise_for_status()
            
            # Trả về dữ liệu JSON dạng Dictionary của Python:
            return response.json()
            
        except requests.exceptions.RequestException as e:
            print(f"Lỗi kết nối mạng: {e}")
            if attempt == max_retries:
                raise Exception("Đã hết số lần thử lại. Pipeline thất bại!")
            time.sleep(2 ** attempt)

# ==========================================
# BƯỚC 2: TRANSFORM - LÀM PHẲNG DỮ LIỆU
# ==========================================
def transform_weather_data(raw_json: dict) -> pd.DataFrame:
    """Làm phẳng cấu trúc JSON lồng nhau và chọn lọc các cột cần thiết."""
    current = raw_json.get("current_weather", {})
    
    # Tạo một DataFrame 1 dòng chứa dữ liệu thời tiết hiện tại:
    record = {
        "location_id": "HCM_DISTRICT_1",
        "recorded_at": current.get("time"),
        "temperature_c": current.get("temperature"),
        "wind_speed": current.get("windspeed"),
        "weather_code": current.get("weathercode")
    }
    df = pd.DataFrame([record])
    
    # Chuẩn hóa kiểu dữ liệu:
    df["recorded_at"] = pd.to_datetime(df["recorded_at"])
    df["temperature_c"] = df["temperature_c"].astype(float)
    return df

# ==========================================
# BƯỚC 3: LOAD - NẠP VÀO DATABASE (UPSERT)
# ==========================================
def load_to_database(df: pd.DataFrame, db_path: str = "weather_warehouse.db"):
    """Nạp dữ liệu vào SQLite với cơ chế UPSERT để đảm bảo tính Idempotent."""
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    
    # 1. Tạo bảng nếu chưa tồn tại (Khóa chính kết hợp: location_id + recorded_at):
    cursor.execute("""
    CREATE TABLE IF NOT EXISTS fact_weather (
        location_id TEXT NOT NULL,
        recorded_at TIMESTAMP NOT NULL,
        temperature_c REAL,
        wind_speed REAL,
        weather_code INTEGER,
        PRIMARY KEY (location_id, recorded_at)
    );
    """)
    
    # 2. Thực hiện lệnh UPSERT: Nếu đã có dòng này thì UPDATE, chưa có thì INSERT:
    upsert_sql = """
    INSERT INTO fact_weather (location_id, recorded_at, temperature_c, wind_speed, weather_code)
    VALUES (?, ?, ?, ?, ?)
    ON CONFLICT(location_id, recorded_at) DO UPDATE SET
        temperature_c = excluded.temperature_c,
        wind_speed = excluded.wind_speed,
        weather_code = excluded.weather_code;
    """
    
    for _, row in df.iterrows():
        cursor.execute(upsert_sql, (
            row["location_id"],
            row["recorded_at"].strftime('%Y-%m-%d %H:%M:%S'),
            row["temperature_c"],
            row["wind_speed"],
            row["weather_code"]
        ))
        
    conn.commit()
    conn.close()
    print("-> Nạp dữ liệu vào Database thành công (Đạt chuẩn Idempotent)!")

# ==========================================
# ĐIỀU PHỐI TOÀN BỘ PIPELINE (MAIN RUN)
# ==========================================
if __name__ == "__main__":
    print("=== BẮT ĐẦU CHẠY PIPELINE ETL TỰ ĐỘNG ===")
    # Tọa độ TP. Hồ Chí Minh: Lat 10.8231, Long 106.6297
    data_json = extract_weather_data(latitude=10.8231, longitude=106.6297)
    df_clean = transform_weather_data(data_json)
    print("\nDữ liệu sau khi làm sạch:")
    print(df_clean)
    
    load_to_database(df_clean)
    print("=== HOÀN TẤT PIPELINE ===")

5. Góc Ôn Luyện Phỏng Vấn (Interview Corner)

❓ Câu hỏi 1: Tính Lũy đẳng (Idempotency) trong thiết kế Data Pipeline là gì? Tại sao một Data Engineer bắt buộc phải đảm bảo tính chất này khi viết script nạp dữ liệu?

Gợi ý trả lời ngắn gọn, chuẩn xác:
- Định nghĩa Idempotency:
  - Là tính chất đảm bảo rằng việc thực thi một tác vụ dữ liệu nhiều lần liên tiếp (với cùng một dữ liệu đầu vào) luôn mang lại cùng một trạng thái dữ liệu duy nhất, không làm biến dạng hay nhân bản dữ liệu rác.
- Tầm quan trọng sống còn:
  - Trong thực tế, các lỗi mạng, lỗi máy chủ hoặc lỗi timeout luôn xảy ra. Khi pipeline bị dừng giữa chừng, hệ thống tự động (như Airflow) sẽ kích hoạt cơ chế chạy lại (Retry).
  - Nếu pipeline không có tính Idempotent (dùng lệnh INSERT thông thường): Việc chạy lại sẽ khiến hàng nghìn bản ghi bị chèn lặp lại lần thứ 2, làm sai lệch doanh thu của toàn bộ báo cáo tài chính.
  - Khi pipeline đảm bảo tính Idempotent (áp dụng kỹ thuật UPSERT hoặc xóa phân vùng trước khi ghi đè INSERT OVERWRITE): Kỹ sư có thể tự tin kích hoạt chạy lại kịch bản bất kỳ lúc nào mà cơ sở dữ liệu vẫn luôn trong trạng thái sạch sẽ tuyệt đối.

❓ Câu hỏi 2: Khi gọi API của đối tác để lấy dữ liệu, bạn nhận được mã lỗi HTTP `429 Too Many Requests`. Nguyên nhân là gì và bạn sẽ thiết kế đoạn mã Python như thế nào để xử lý an toàn mà không làm sập tiến trình?

Gợi ý trả lời:
- Nguyên nhân:
  - Bạn đã vi phạm chính sách giới hạn tần suất (Rate Limit) của nhà cung cấp API (ví dụ: họ chỉ cho phép gọi tối đa $60\text{ requests/phút}$ , nhưng script của bạn dùng vòng lặp gửi hàng trăm request chỉ trong $2\text{ giây}$ ). Máy chủ đối tác chủ động chặn bạn để bảo vệ hệ thống của họ.
- Chiến lược xử lý bằng Python:
  1. Kiểm tra Header Retry-After: Nhiều API chuyên nghiệp sẽ trả về kèm header Retry-After: 30 (cho biết bạn cần nghỉ đúng $30\text{ giây}$ trước khi được phép gọi lại).
  2. Áp dụng thuật toán Exponential Backoff kết hợp Jitter (Độ trễ ngẫu nhiên):
    - Tạm dừng chương trình bằng hàm time.sleep() với thời gian chờ tăng dần theo cấp số nhân ( $2^n$ giây) cộng thêm một số giây ngẫu nhiên nhỏ (Jitter) để phân tán thời điểm gửi lại.
  3. Chủ động đặt nhịp độ (Rate Limiting nội bộ): Nếu cần gọi danh sách $1,000$ sản phẩm, chủ động đặt lệnh time.sleep(1.0) sau mỗi vòng lặp để giữ tốc độ dưới ngưỡng trần quy định của đối tác.

❓ Câu hỏi 3: Khi nhận được một tập dữ liệu JSON từ API có nhiều cấp độ lồng nhau (ví dụ một khách hàng chứa danh sách mảng nhiều đơn hàng), làm thế nào để chuyển đổi nó thành một Pandas DataFrame phẳng mà không phải viết các vòng lặp `for` thủ công phức tạp?

Gợi ý trả lời:
- Sử dụng hàm chuyên dụng pd.json_normalize() của Pandas:
  - Thay vì viết nhiều vòng lặp for lồng nhau chậm chạp và dễ lỗi KeyError, Pandas cung cấp sẵn hàm pd.json_normalize().
- Cách thức hoạt động:
  - Tự động làm phẳng các từ điển con (Nested Dictionaries) bằng cách ghép tên khóa cha và khóa con qua dấu chấm: address.city, address.zipcode.
  - Với các trường là mảng danh sách lồng nhau (Nested Lists, ví dụ: mảng orders), sử dụng tham số record_path=['orders'] và meta=['user_id', 'user_name']:
    - Hàm sẽ tự động "nhân bản" thông tin người dùng ở cấp cha để ghép nối tương ứng với từng dòng đơn hàng con ở cấp dưới.
  - Kết quả trả về trực tiếp là một DataFrame 2 chiều chuẩn chỉnh, sẵn sàng cho việc làm sạch hoặc nạp thẳng vào cơ sở dữ liệu quan hệ.

6. Tóm Tắt Ghi Nhớ Nhanh (Key Takeaways)

Chu trình Ingestion chuẩn: Gọi API (requests) $\rightarrow$ Làm phẳng JSON (json_normalize) $\rightarrow$ Nạp DB (Upsert).
Luôn đặt tham số timeout trong requests.get() để ngăn chặn việc script bị treo vĩnh viễn khi mạng đối tác bị đơ.
Khi gặp lỗi mạng hoặc mã 429 / 5xx, luôn áp dụng Exponential Backoff (chờ tăng dần $2^n$ giây) trước khi thử lại.
Idempotency là nguyên tắc sống còn số 1 của Data Pipeline: Luôn dùng kỹ thuật UPSERT (ON CONFLICT DO UPDATE) thay cho INSERT thuần túy.
Ưu tiên lấy dữ liệu qua REST API chính thức; chỉ dùng Web Scraping khi đối tác hoàn toàn không cung cấp cổng kết nối.