Python DE 8 — Chất lượng code & production

Khép lại series: từ "script chạy được" đến "pipeline đáng tin cậy"

Suốt bảy bài trước, chúng ta đã đi từ tổng quan Python cho DE, qua pandas, Polars/Arrow, file formats, kết nối database, gọi API/ingestion, tới đồng thời & hiệu năng. Tất cả cho bạn công cụ để làm ra dữ liệu. Bài cuối này trả lời câu hỏi khác hẳn: làm sao để cái pipeline đó chạy lại được, kiểm thử được, quan sát được và đóng gói được — tức là đủ điều kiện lên production.

Ranh giới giữa một script và một pipeline production không nằm ở độ phức tạp của thuật toán. Nó nằm ở những thứ tưởng chừng "phụ": có test không, log có đọc được không, config có hardcode secret không, và khi lỗi lúc 2 giờ sáng thì có tự retry hay chết im lặng. Một script 50 dòng làm đúng việc vẫn có thể là gánh nặng vận hành; một pipeline có kỷ luật lại là tài sản.

Như thường lệ: sandbox của khoá là PostgreSQL, Python không chạy trực tiếp ở đây. Mọi khối mã dưới đây là (minh hoạ) — bạn đọc để hiểu cấu trúc và ý đồ, rồi mang về dự án thật để chạy.

Sơ đồ trên là bản đồ của cả bài. Mỗi mũi tên là một mức trưởng thành, và mũi tên đứt nét nhắc rằng production không phải điểm cuối — nó vòng lại thành yêu cầu tiến hoá tiếp cho code.

1. Cấu trúc project: nơi mọi thứ có chỗ đứng

Script khởi đầu thường là một file pipeline.py chứa cả đọc dữ liệu, biến đổi và ghi. Bước trưởng thành đầu tiên là dựng một layout chuẩn theo mô hình src/ layout:

myde/
├── pyproject.toml         # khai báo project + dependency
├── uv.lock                # (hoặc poetry.lock) khoá phiên bản chính xác
├── README.md
├── src/
│   └── myde/
│       ├── __init__.py
│       ├── config.py      # settings qua pydantic
│       ├── io.py          # đọc/ghi: DB, S3, file — chỗ chạm thế giới ngoài
│       ├── transform.py   # logic thuần: input -> output, KHÔNG side effect
│       └── pipeline.py    # ghép io + transform lại
└── tests/
    ├── test_transform.py
    └── test_pipeline.py

Lý do dùng src/ layout không phải thẩm mỹ: nó ép bạn cài package trước khi test, nên test chạy trên đúng thứ production dùng, thay vì chạy trên file cạnh nó và che giấu lỗi import.

Nguyên tắc kiến trúc quan trọng nhất ở đây: tách I/O khỏi logic. Hàm biến đổi (transform.py) nhận vào một DataFrame/dict và trả ra một DataFrame/dict — không tự mở kết nối, không đọc file, không gọi API. Mọi chuyện "chạm thế giới ngoài" dồn vào io.py. Vì sao? Vì logic thuần thì test cực dễ (đưa input, so output), còn I/O thì test chậm và giòn. Ranh giới này chính là điều làm phần 3 (pytest) trở nên khả thi.

# transform.py (minh hoạ) — thuần, không side effect
import pandas as pd

def normalize_customers(df: pd.DataFrame) -> pd.DataFrame:
    out = df.copy()
    out["email"] = out["email"].str.strip().str.lower()
    out = out.dropna(subset=["customer_id"])
    out = out.drop_duplicates(subset=["customer_id"], keep="last")
    return out

2. Packaging & môi trường: pyproject.toml + lockfile

"Trên máy tôi chạy được" là câu nói tai tiếng nhất ngành. Gốc rễ của nó là môi trường không tái lập được. Giải pháp: khai báo dependency trong pyproject.toml và khoá phiên bản chính xác bằng lockfile.

# pyproject.toml (minh hoạ)
[project]
name = "myde"
version = "0.1.0"
requires-python = ">=3.11"
dependencies = [
    "pandas>=2.2",
    "pydantic>=2.7",
    "pydantic-settings>=2.3",
    "sqlalchemy>=2.0",
]

[project.optional-dependencies]
dev = ["pytest>=8", "mypy>=1.10", "ruff>=0.5", "pre-commit>=3.7"]

[project.scripts]
myde = "myde.pipeline:main"   # tạo lệnh CLI `myde`

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[tool.ruff]
line-length = 100

[tool.mypy]
python_version = "3.11"
strict = true

Về công cụ quản lý môi trường, hai lựa chọn phổ biến hiện nay:

• uv — trình quản lý viết bằng Rust, rất nhanh; sinh uv.lock. Lệnh điển hình: uv sync (dựng môi trường đúng theo lock), uv add pandas, uv run pytest.
• poetry — lâu đời, sinh poetry.lock; lệnh poetry install, poetry add, poetry run.

Điểm chung — và là điều bắt buộc cho production — là commit lockfile vào git. pyproject.toml nói "tôi cần pandas >= 2.2"; lockfile nói "chính xác pandas 2.2.3 với đúng các hash này". CI, Docker và máy đồng nghiệp đều cài từ lockfile nên ra kết quả giống hệt nhau. [project.scripts] cho bạn một entry point CLI gọn (myde) thay vì phải nhớ đường dẫn file — rất tiện khi đóng Docker.

3. Type hints & pydantic: hợp đồng dữ liệu rõ ràng

Type hints không làm Python chạy nhanh hơn, nhưng làm code rõ ý định và cho phép mypy bắt lỗi trước khi chạy — thứ vô giá với code dữ liệu, nơi một cột đổi kiểu có thể phá cả pipeline im lặng.

Với mypy --strict, mọi hàm phải có annotation. Bạn không cần chạy Python để hưởng lợi: mypy phân tích tĩnh và báo "hàm này nhận str nhưng bạn đưa int" ngay trong CI.

pydantic đóng vai trò khác: validate dữ liệu lúc chạy. Type hint được mypy kiểm khi build; pydantic kiểm khi dữ liệu thật chảy qua. Rất hợp để kiểm định bản ghi ingest từ API/DB — nơi bạn không kiểm soát nguồn.

# validate bản ghi ingest (minh hoạ)
from datetime import date
from pydantic import BaseModel, EmailStr, field_validator

class Customer(BaseModel):
    customer_id: int
    email: EmailStr
    signup_date: date

    @field_validator("customer_id")
    @classmethod
    def positive_id(cls, v: int) -> int:
        if v <= 0:
            raise ValueError("customer_id phải dương")
        return v

# Customer(**row) sẽ raise ValidationError nếu dữ liệu bẩn
# -> bạn bắt được ngay tại biên ingest thay vì để nó ngấm vào kho

Chiến lược thực dụng: validate tại biên (khi dữ liệu vừa vào từ nguồn ngoài) rồi bên trong pipeline tin tưởng kiểu đã sạch. Đừng validate lại ở mọi hàm — vừa chậm vừa rối.

4. Kiểm thử với pytest: test logic, mock I/O

Nhờ đã tách I/O khỏi logic ở phần 1, phần lớn giá trị test nằm ở unit test cho transform: nhanh, tất định, không cần mạng hay DB.

# tests/test_transform.py (minh hoạ)
import pandas as pd
import pytest
from myde.transform import normalize_customers

@pytest.fixture
def raw() -> pd.DataFrame:
    return pd.DataFrame({
        "customer_id": [1, 2, 2, None],
        "email": ["  [email protected] ", "[email protected]", "[email protected]", "[email protected]"],
    })

def test_lowercases_and_trims_email(raw):
    out = normalize_customers(raw)
    assert out.loc[out.customer_id == 1, "email"].iloc[0] == "[email protected]"

def test_drops_null_id_and_dedupes(raw):
    out = normalize_customers(raw)
    assert out["customer_id"].tolist() == [1, 2]

@pytest.mark.parametrize("bad_email", ["", "no-at-sign", None])
def test_invalid_email_rejected(bad_email):
    from pydantic import ValidationError
    from myde.config import Customer  # ví dụ
    with pytest.raises(ValidationError):
        Customer(customer_id=1, email=bad_email, signup_date="2026-01-01")

Ba công cụ pytest đáng nắm chắc:

• fixtures — dữ liệu/tài nguyên dùng chung, dựng sạch cho từng test.
• parametrize — chạy cùng logic với nhiều bộ input, tránh copy-paste test.
• mock — thay I/O thật bằng giả để test không phụ thuộc mạng/DB.

Ví dụ mock một lời gọi API để test logic ingest mà không thật sự ra internet:

# tests/test_ingest.py (minh hoạ) — mock API
from unittest.mock import patch
from myde.io import fetch_orders

def test_fetch_orders_parses_payload():
    fake = {"orders": [{"id": 1, "total": 10.0}]}
    with patch("myde.io.requests.get") as m:
        m.return_value.json.return_value = fake
        m.return_value.raise_for_status.return_value = None
        orders = fetch_orders(customer_id=1)
    assert orders[0]["total"] == 10.0

Ngoài test logic, code dữ liệu cần test chất lượng dữ liệu: sau khi transform, khẳng định các bất biến (không null ở khoá, số dòng trong khoảng hợp lý, tổng doanh thu không âm). Đây là tư duy giống test trong dbt nhưng viết bằng pytest ngay trong code Python:

def test_output_invariants(raw):
    out = normalize_customers(raw)
    assert out["customer_id"].notna().all()
    assert out["customer_id"].is_unique
    assert len(out) <= len(raw)

5. Logging & cấu hình: quan sát được, không hardcode secret

Bỏ print. print không có mức độ (level), không có timestamp, không tắt bật được, và trên K8s thì lẫn lộn với stdout khác. Dùng module logging chuẩn, và trong môi trường container hãy log JSON có cấu trúc để hệ thống thu log (Loki, ELK, Cloud Logging) phân tích được từng trường.

# logging JSON có cấu trúc (minh hoạ)
import json, logging, sys

class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        payload = {
            "level": record.levelname,
            "logger": record.name,
            "msg": record.getMessage(),
            "time": self.formatTime(record),
        }
        if record.exc_info:
            payload["exc"] = self.formatException(record.exc_info)
        return json.dumps(payload)

handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(JsonFormatter())
logging.basicConfig(level=logging.INFO, handlers=[handler])
log = logging.getLogger("myde")

log.info("bắt đầu load", extra={"rows": 12000})

Về cấu hình: không hardcode đường dẫn DB, mật khẩu, endpoint. Đọc từ biến môi trường, và pydantic-settings biến việc này thành hợp đồng có kiểu, có validate:

# config.py (minh hoạ) — pydantic-settings
from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_prefix="MYDE_", env_file=".env")

    db_url: str = Field(..., description="chuỗi kết nối Postgres")
    batch_size: int = 5000
    log_level: str = "INFO"

settings = Settings()   # đọc MYDE_DB_URL, MYDE_BATCH_SIZE... từ env

Điểm mấu chốt: secret (mật khẩu DB, token API) đi vào qua env/secret manager, không nằm trong git, không nằm trong image Docker. .env chỉ dùng cho máy dev và phải nằm trong .gitignore. Trên K8s, chúng đến từ Secret; trên Airflow, từ Connections/Variables.

6. Xử lý lỗi, retry idempotent & dead-letter

Pipeline production sẽ gặp lỗi tạm thời: API timeout, DB rớt kết nối, mạng chập chờn. Ba nguyên tắc:

• Retry có backoff cho lỗi tạm thời — thử lại vài lần với khoảng chờ tăng dần. Chỉ retry lỗi có khả năng tự khỏi (timeout, 503), đừng retry lỗi logic (400, dữ liệu sai) vì retry chỉ lặp lại thất bại.
• Idempotent — chạy lại phải ra cùng kết quả, không nhân đôi dữ liệu. Cách thường dùng: ghi theo kiểu upsert/MERGE theo khoá, hoặc xoá-rồi-ghi theo partition (ghi đè ngày 2026-06-30 thay vì chèn thêm). Đây là điều kiện sống còn để Airflow retry an toàn.
• Dead-letter — bản ghi bẩn không được chặn cả lô. Tách chúng ra một nơi riêng (bảng/thư mục "quarantine") để xử lý sau, còn phần sạch vẫn chảy tiếp.

# retry đơn giản có backoff (minh hoạ)
import time
from typing import Callable, TypeVar
T = TypeVar("T")

def with_retry(fn: Callable[[], T], attempts: int = 3, base: float = 1.0) -> T:
    for i in range(attempts):
        try:
            return fn()
        except (TimeoutError, ConnectionError) as e:
            if i == attempts - 1:
                raise
            log.warning("retry", extra={"attempt": i + 1, "err": str(e)})
            time.sleep(base * 2 ** i)   # 1s, 2s, 4s
    raise RuntimeError("unreachable")

7. Lint & format tự động: ruff + pre-commit

Tranh luận về dấu cách và thứ tự import là lãng phí thời gian con người. Giao cho máy. ruff là công cụ (viết bằng Rust) làm cả lint (bắt lỗi phong cách/logic nhẹ) lẫn format, rất nhanh, thay được cho cả flake8 + isort + black.

• ruff check . — lint, phát hiện import thừa, biến không dùng, lỗi thường gặp.
• ruff format . — format lại code đồng nhất.

pre-commit chạy các kiểm tra này tự động ngay trước mỗi commit, nên code bẩn không bao giờ lọt vào git:

# .pre-commit-config.yaml (minh hoạ)
repos:
  - repo: https://github.com/astral-sh/ruff-pre-commit
    rev: v0.5.0
    hooks:
      - id: ruff
        args: [--fix]
      - id: ruff-format
  - repo: https://github.com/pre-commit/mirrors-mypy
    rev: v1.10.0
    hooks:
      - id: mypy

Cùng những kiểm tra này lặp lại trong CI (phần 9), tạo thành lưới an toàn hai lớp: dev bắt sớm, CI bắt chắc.

8. Đóng gói Docker: image gọn cho job Python

Để chạy đồng nhất trên Airflow hay K8s, đóng gói pipeline thành một image Docker. Nếu bạn chưa quen Dockerfile, hãy đọc K8s 2 — Dockerfile trước. Mục tiêu ở đây là image gọn và cài từ lockfile:

# Dockerfile (minh hoạ) — multi-stage, image gọn
FROM python:3.11-slim AS base
ENV PYTHONUNBUFFERED=1 PIP_NO_CACHE_DIR=1

# cài dependency trước (tận dụng layer cache)
WORKDIR /app
COPY pyproject.toml uv.lock ./
RUN pip install --no-cache-dir uv && uv sync --frozen --no-dev

# copy code sau cùng
COPY src/ ./src/
RUN uv pip install --no-deps -e .

# chạy như user thường, không root
RUN useradd -m app
USER app

ENTRYPOINT ["myde"]   # gọi entry point đã khai báo trong pyproject

Bốn nguyên tắc gói image DE gọn và an toàn:

• base -slim (hoặc distroless) — nhỏ hơn nhiều so với image đầy đủ.
• --frozen / cài từ lock — đảm bảo image dùng đúng phiên bản đã khoá, không "trôi" theo thời gian build.
• cache layer đúng thứ tự — copy dependency trước, code sau; đổi code không phải cài lại thư viện.
• chạy non-root — nguyên tắc an toàn tối thiểu trên K8s.

Nếu bạn còn nhớ mẹo trong ghi chú "Docker build sau proxy CA": khi build sau proxy công ty phải trust CA của công ty, nếu không pip/uv sẽ lỗi SSL.

9. CI/CD: lint + test trên PR, build image khi merge

Bước cuối để "chất lượng" không phụ thuộc trí nhớ con người là tự động hoá trên CI. Mẫu tối thiểu: mọi PR chạy lint + type check + test; khi merge vào nhánh chính thì build và đẩy image.

# .github/workflows/ci.yml (minh hoạ)
name: ci
on: [pull_request, push]
jobs:
  quality:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: astral-sh/setup-uv@v3
      - run: uv sync --frozen
      - run: uv run ruff check .
      - run: uv run ruff format --check .
      - run: uv run mypy src/
      - run: uv run pytest -q

Ý tưởng: CI cài từ lockfile (--frozen) nên tái lập chính xác môi trường, rồi chạy đúng chuỗi kiểm tra như pre-commit. PR nào đỏ thì không merge. Job build image (bước riêng, chạy khi push vào main) đóng gói và đẩy lên registry để Airflow/K8s kéo về.

Use case thực tế

Một job ingest đơn hàng hằng ngày, ban đầu là daily_orders.py một file: gọi API, biến đổi bằng pandas, to_sql vào Postgres, print vài dòng tiến độ, mật khẩu DB nằm ngay trong code. Nó chạy được vài tuần rồi bắt đầu rắc rối: một hôm API đổi định dạng một trường và job ghi dữ liệu sai vào kho mà không ai biết; một hôm khác job chạy lại sau lỗi mạng và nhân đôi đơn hàng.

Áp dụng bài này, đội DE tái cấu trúc: tách fetch_orders/write_orders (I/O) khỏi transform_orders (logic thuần); dùng pydantic validate từng đơn ngay tại biên ingest (đơn bẩn rơi vào bảng orders_quarantine thay vì phá cả lô); viết pytest cho transform và cho các bất biến chất lượng dữ liệu; đổi to_sql chèn-thêm thành upsert theo order_id để retry idempotent; chuyển print sang logging JSON và đọc MYDE_DB_URL qua pydantic-settings. Cuối cùng đóng Docker image gọn, để Airflow chạy trên K8s gọi image đó, và bật CI chạy ruff + mypy + pytest trên mỗi PR.

Kết quả không phải "chạy nhanh hơn" — mà là khi API đổi định dạng lần sau, test đỏ ngay trên PR trước khi lên production, và khi mạng chập chờn thì retry không còn nhân đôi dữ liệu. Đó chính là khác biệt giữa script và pipeline.

Ghi nhớ

• Tách I/O khỏi logic là quyết định kiến trúc quan trọng nhất — nó làm test khả thi và code dễ đọc.
• Type hints (mypy) kiểm lúc build; pydantic kiểm lúc chạy. Validate dữ liệu ngoài ngay tại biên ingest.
• Không print, không hardcode secret. Logging JSON có cấu trúc + config qua env/pydantic-settings.
• Lockfile là nguồn chân lý của môi trường. Dev, CI, Docker cùng cài từ lock nên tái lập được.
• Retry chỉ cho lỗi tạm thời; ghi phải idempotent; dữ liệu bẩn vào dead-letter, không chặn cả lô.
• Tự động hoá chất lượng: ruff + pre-commit ở local, lint/mypy/pytest ở CI. Đừng dựa vào trí nhớ.

Checklist production

• src/ layout; I/O tách khỏi logic thuần.
• pyproject.toml + lockfile (uv.lock/poetry.lock) đã commit vào git.
• Type hints đầy đủ; mypy sạch trong CI.
• pydantic validate dữ liệu tại biên; có xử lý bản ghi bẩn (dead-letter).
• pytest cho transform + test bất biến chất lượng dữ liệu; mock cho API/DB.
• Logging có cấu trúc (JSON trên container), không còn print.
• Config qua env/pydantic-settings; không secret trong git hay image.
• Retry có backoff cho lỗi tạm thời; ghi idempotent (upsert / ghi đè partition).
• ruff (lint + format) + pre-commit đã bật.
• Dockerfile gọn (-slim, multi-stage, non-root, cài từ lock).
• CI chạy lint + mypy + pytest trên mỗi PR; build image khi merge.

Đến đây series Python cho Data Engineering khép lại. Bạn đã có công cụ để làm ra dữ liệu, và giờ có kỷ luật để đưa nó lên production một cách đáng tin cậy. Chúc bạn xây được những pipeline mà nửa đêm không phải thức dậy vì nó.