Python DE 7 — Đồng thời, song song & hiệu năng

Vì sao một DE cần hiểu chuyện này?

Data Engineering là nghề "bơm" dữ liệu: kéo hàng trăm file, gọi vài trăm endpoint API, đọc bảng vài chục triệu dòng rồi biến đổi. Sớm muộn bạn cũng gặp một job chạy 40 phút mà lẽ ra chỉ cần 2 phút. Lúc đó câu hỏi không phải "Python có nhanh không" mà là "nút thắt nằm ở đâu, và công cụ nào phá được nó".

Bài này trả lời ba câu hỏi thực dụng: (1) vì sao Python có tiếng "chậm" và cái GIL liên quan thế nào; (2) khi nào dùng threading, asyncio, multiprocessing, hay vector hoá; (3) làm sao đo để biết chắc chỗ nào đáng tối ưu thay vì đoán mò. Toàn bộ ví dụ là minh hoạ — sandbox của khoá là PostgreSQL, Python không chạy trực tiếp ở đây.

Nguyên tắc xuyên suốt, hãy khắc cốt ghi tâm ngay từ đầu: đo trước, sửa nút thắt lớn nhất, đừng tối ưu sớm. Rất nhiều lần lời giải tốt nhất không phải "song song hoá vòng lặp Python" mà là "đổi công cụ" — dùng Polars, DuckDB hay Spark thay vì gồng ép CPython.

1. GIL — vì sao một tiến trình Python không chạy song song thật

GIL (Global Interpreter Lock) là một khoá toàn cục trong CPython (bản Python phổ biến nhất). Quy tắc của nó rất đơn giản và rất phiền: tại một thời điểm, chỉ một thread được phép thực thi bytecode Python. Bạn có thể tạo 10 thread, nhưng nếu cả 10 đều đang tính toán thuần Python, chúng vẫn thay phiên nhau chạy trên đúng một lõi — không có song song thật.

Vì sao CPython làm vậy? Vì bộ đếm tham chiếu (reference counting) để quản lý bộ nhớ không an toàn khi nhiều thread cùng sửa. GIL là cách rẻ để đảm bảo an toàn mà không phải khoá từng đối tượng. Cái giá phải trả: các tác vụ CPU-bound không tận dụng được nhiều lõi trong một tiến trình.

Nhưng — điểm mấu chốt cho DE — GIL được nhả ra trong lúc chờ I/O. Khi một thread gọi mạng, đọc đĩa, hay time.sleep, nó nhả GIL cho thread khác chạy. Vì vậy threading rất hữu ích cho I/O-bound, chỉ vô dụng cho CPU-bound.

Thêm hai ngoại lệ quan trọng:

• Các thư viện C nhả GIL. NumPy, pandas, Polars, scikit-learn... làm phần nặng bằng C/Rust và nhả GIL trong lúc tính. Nghĩa là một phép nhân ma trận NumPy có thể chạy song song trên nhiều lõi ngay cả khi bạn ở trong một tiến trình Python. Đây là lý do vector hoá thường thắng.
• No-GIL đang tới. PEP 703 giới thiệu bản build "free-threaded" (--disable-gil), thử nghiệm trong Python 3.13 (2024) và được nâng lên trạng thái supported (chưa mặc định) trong Python 3.14 theo PEP 779. Khi ổn định, thread thuần Python sẽ chạy song song thật. Hiện tại (2026) nó vẫn là lựa chọn build riêng, chưa nên dựa vào cho production.

Tóm gọn: CPU-bound + Python thuần + threading = vô ích. I/O-bound + threading = tốt. CPU-bound thật sự cần multiprocessing hoặc thư viện C.

2. Phân loại tác vụ: I/O-bound hay CPU-bound?

Trước khi chọn công cụ, phải trả lời một câu: thời gian đang trôi vào việc chờ hay việc tính?

• I/O-bound — phần lớn thời gian là chờ: chờ phản hồi API, chờ query DB, chờ đọc/ghi file, chờ mạng. CPU gần như rảnh rỗi. Đây là kịch bản DE gặp nhiều nhất. Xem thêm bài API & async.
• CPU-bound — phần lớn thời gian là tính: parse/transform hàng triệu bản ghi bằng Python thuần, nén/giải nén, tính toán số học, JSON lớn. CPU cháy 100%, không ai chờ ai.

Cách nhận biết nhanh: mở htop lúc job chạy. CPU một lõi cắm 100% → CPU-bound. CPU nhàn nhưng job vẫn lâu → I/O-bound.

Từ đó suy ra công cụ:

Ba lựa chọn concurrency đều nằm gọn trong thư viện chuẩn: concurrent.futures cho hai loại pool, và asyncio cho vòng lặp sự kiện.

3. Cây quyết định

Đọc cây này từ trên xuống mỗi khi gặp job chậm. Chú ý nhánh trái cùng: nếu chưa đo, đừng chọn công cụ vội — bạn có thể đang tối ưu nhầm chỗ.

4. threading cho I/O-bound

Kịch bản kinh điển của DE: tải song song nhiều file/URL. ThreadPoolExecutor là cách gọn nhất — bạn viết hàm xử lý một item, executor lo phần chạy song song.

# (minh hoạ) tải song song nhiều URL bằng ThreadPoolExecutor
from concurrent.futures import ThreadPoolExecutor, as_completed
import requests

def fetch(url: str) -> tuple[str, int]:
    r = requests.get(url, timeout=30)
    return url, len(r.content)

urls = [f"https://api.example.com/data/{i}" for i in range(500)]

results = []
with ThreadPoolExecutor(max_workers=32) as pool:
    futures = {pool.submit(fetch, u): u for u in urls}
    for fut in as_completed(futures):
        url = futures[fut]
        try:
            results.append(fut.result())
        except Exception as e:
            print(f"Lỗi {url}: {e}")

Điểm cần nhớ:

• max_workers không phải càng nhiều càng tốt. Mỗi thread tốn 1-8 MB RAM và có chi phí chuyển ngữ cảnh. Với I/O, 16-64 thường là dải hợp lý; đo để chỉnh.
• Dùng as_completed để xử lý kết quả ngay khi có, thay vì chờ tất cả.
• Luôn bọc try/except quanh fut.result(): một item lỗi không được làm sập cả job.
• Đặt timeout cho mọi lời gọi mạng — không có timeout là công thức treo job.

Nếu bạn cần hàng nghìn kết nối đồng thời, thread trở nên tốn kém (RAM + context switch); lúc đó asyncio là lựa chọn đúng. Cơ chế async/await và httpx đã được trình bày ở bài API & async — về bản chất asyncio làm cùng việc (chờ I/O đồng thời) nhưng trên một thread duy nhất với overhead nhỏ hơn nhiều.

5. multiprocessing cho CPU-bound

Khi hotspot là tính toán Python thuần (và không vector hoá được), cách vượt GIL là tạo nhiều tiến trình — mỗi tiến trình có interpreter và GIL riêng, chạy thật trên các lõi khác nhau. API gần như y hệt, chỉ đổi ThreadPoolExecutor thành ProcessPoolExecutor.

# (minh hoạ) tính toán nặng chia cho nhiều tiến trình
from concurrent.futures import ProcessPoolExecutor
import os

def heavy_transform(record: dict) -> dict:
    # giả lập logic parse/tính toán nặng bằng Python thuần
    total = sum(i * i for i in range(record["n"]))
    return {"id": record["id"], "score": total}

records = [{"id": i, "n": 200_000} for i in range(64)]

with ProcessPoolExecutor(max_workers=os.cpu_count()) as pool:
    results = list(pool.map(heavy_transform, records))

Cảnh báo quan trọng với multiprocessing — đây là chỗ hay "sập bẫy":

• Serialization (pickle) tốn kém. Dữ liệu vào/ra tiến trình con phải được pickle rồi copy. Nếu mỗi call chỉ tính nhẹ mà truyền qua lại cả DataFrame lớn, chi phí copy nuốt hết lợi ích song song. Chỉ đẩy sang process khi việc tính đủ nặng để bù cho chi phí này.
• Startup chậm. Tạo tiến trình mất mili-giây (thread là micro-giây). Với vô số item nhỏ, hãy chia lô (chunk) để mỗi tiến trình làm một mẻ lớn.
• Đặt max_workers = os.cpu_count() là điểm khởi đầu hợp lý; nhiều hơn số lõi thường không giúp cho CPU-bound.
• Trên nền tảng dùng spawn, code phải để trong if __name__ == "__main__": để tránh tạo tiến trình đệ quy.

6. Vector hoá — thường là lời giải tốt hơn cả song song

Đây là bài học đắt giá nhất của bài viết: với dữ liệu dạng bảng, đừng song song hoá vòng lặp Python — hãy xoá vòng lặp đi. Một vòng for trên vài triệu dòng bằng Python thuần vốn đã chậm; chia nó cho 8 tiến trình chỉ nhanh gấp ~8, trong khi viết lại bằng NumPy/pandas/Polars có thể nhanh gấp 50-100 lần vì phép tính chạy trong C và tự nhả GIL.

# (minh hoạ) vòng lặp thuần (chậm) vs vector hoá (nhanh)
import numpy as np

# CÁCH CHẬM: vòng lặp Python thuần
def slow_normalize(values):
    out = []
    mean = sum(values) / len(values)
    for v in values:
        out.append((v - mean) / 100)
    return out

# CÁCH NHANH: vector hoá, chạy trong C, nhả GIL
def fast_normalize(arr: np.ndarray) -> np.ndarray:
    return (arr - arr.mean()) / 100

Thứ tự ưu tiên khi tối ưu một biến đổi dữ liệu:

1. Vector hoá bằng NumPy/pandas — xem pandas chuyên sâu.
2. Nếu dữ liệu lớn hơn RAM hoặc pandas vẫn chậm → Polars (Rust, đa lõi mặc định, lazy) — xem Polars & Arrow.
3. Nếu vẫn không đủ → DuckDB (SQL trên file) hoặc Spark cho quy mô cluster.

Chỉ khi logic thực sự không vector hoá được (parse chuỗi phức tạp, gọi thư viện Python thuần từng dòng) thì multiprocessing mới là câu trả lời.

7. Profiling — đo trước khi tối ưu

Nguyên tắc vàng: bạn không giỏi đoán hotspot đâu. Trực giác về "chỗ chậm" sai thường xuyên đến mức đáng ngại. Hãy đo. Quy luật 80/20 gần như luôn đúng: ~80% thời gian nằm ở ~20% code — tìm cái 20% đó và chỉ sửa nó.

timeit — đo nhanh một đoạn nhỏ, chạy nhiều lần lấy trung bình. Dùng để so hai cách viết.

cProfile — profiler cấp hàm, có sẵn trong thư viện chuẩn. Cho biết mỗi hàm được gọi bao nhiêu lần và tốn bao nhiêu thời gian tích luỹ. Đây là điểm bắt đầu tiêu chuẩn.

# (minh hoạ) cProfile: tìm hàm nào ngốn thời gian nhất
import cProfile
import pstats

def pipeline():
    ...  # toàn bộ job của bạn

profiler = cProfile.Profile()
profiler.enable()
pipeline()
profiler.disable()

stats = pstats.Stats(profiler)
stats.sort_stats("cumulative").print_stats(15)  # 15 hàm nặng nhất

line_profiler — sau khi cProfile chỉ ra hàm nghi vấn, công cụ này cho biết dòng nào trong hàm đó tốn thời gian (trang trí bằng @profile). Chi tiết hơn nhưng overhead cao hơn.

py-spy — profiler dạng sampling, viết bằng Rust, chạy ngoài tiến trình Python nên overhead cực thấp và an toàn cho production. Bạn attach vào một job đang chạy: py-spy top --pid 12345 xem live, hay py-spy record -o profile.svg --pid 12345 xuất flame graph. Đây là công cụ đi tìm "tại sao job production chậm" mà không cần dừng nó.

Quy trình chuẩn: cProfile khoanh vùng hàm → line_profiler soi dòng → py-spy cho production. Đo, sửa đúng một chỗ nặng nhất, rồi đo lại.

8. Bộ nhớ — nút thắt vô hình

RAM cũng là hiệu năng: hết RAM là MemoryError hoặc swap khiến job chậm gấp trăm lần. Với DE, mấu chốt là đừng giữ toàn bộ dữ liệu trong bộ nhớ khi không cần.

Generator/iterator (lazy) — thay vì đọc cả file vào một list, xử lý từng phần và chỉ giữ những gì đang dùng:

# (minh hoạ) generator: đọc & xử lý file lớn theo dòng, RAM gần như không đổi
def parse_large_file(path: str):
    with open(path) as f:
        for line in f:           # đọc lười, không nạp cả file
            yield transform(line)

# tiêu thụ theo luồng, không dựng list khổng lồ
for record in parse_large_file("huge.jsonl"):
    write_to_sink(record)

Xử lý theo chunk/stream — với bảng lớn hơn RAM, đọc theo lô (pandas.read_csv(..., chunksize=...), hoặc chế độ lazy/streaming của Polars) rồi ghi từng lô ra sink. Không bao giờ để cả dataset nằm cùng lúc trong RAM.

memory_profiler — đo mức RAM theo từng dòng để phát hiện chỗ phình bộ nhớ hoặc rò rỉ (giữ tham chiếu không cần thiết).

Một mẹo dtype: mảng số đồng nhất (NumPy/Arrow) tốn ít RAM hơn hẳn list các object Python, vì object Python mang theo overhead lớn mỗi phần tử. Chuyển dữ liệu sang dtype gọn (int32 thay int64, category cho chuỗi lặp) thường giảm RAM 2-4 lần — chi tiết ở bài pandas chuyên sâu.

9. Vài mẹo tăng tốc khác

• Caching với functools.lru_cache — nếu một hàm thuần (cùng đầu vào → cùng đầu ra) được gọi lặp lại với cùng tham số, cache kết quả. Rất hợp cho tra cứu, chuẩn hoá lặp, gọi API idempotent.

# (minh hoạ) cache kết quả hàm đắt đỏ
from functools import lru_cache

@lru_cache(maxsize=10_000)
def lookup_region(zip_code: str) -> str:
    return expensive_geo_lookup(zip_code)

• Batch I/O — ghi DB 1000 dòng một lần (executemany / COPY) thay vì 1000 lần insert; đọc/ghi theo lô lớn giảm số vòng round-trip.
• Giảm serialization — mỗi lần dữ liệu qua ranh giới tiến trình/mạng đều tốn pickle/JSON. Truyền ít, truyền dạng cột (Arrow/Parquet) thay vì hàng.
• numba / Cython — nếu buộc phải giữ vòng lặp số học Python và đã vắt kiệt vector hoá, numba biên dịch JIT hàm số học sang mã máy (thêm một decorator), Cython dịch sang C. Đây là biện pháp cuối, dùng có chọn lọc cho đúng hotspot.

Use case thực tế

(1) Tải song song 500 file/API — I/O-bound. Một pipeline daily cần kéo 500 file JSON từ một REST API. Chạy tuần tự: mỗi request ~400 ms → hơn 3 phút, mà CPU nằm không suốt thời gian đó. Đây rõ ràng là I/O-bound. Giải pháp: ThreadPoolExecutor(max_workers=32) như ở Mục 4 → thời gian rơi xuống ~10-15 giây vì các request chờ đồng thời. Nếu quy mô lên hàng nghìn/chục nghìn endpoint, chuyển sang asyncio + httpx (bài API & async) để không phải nuôi hàng nghìn thread. Luôn thêm timeout + retry + giới hạn concurrency để không "dập" server nguồn.

(2) Xử lý CPU nặng — process hay vectorize? Cần chuẩn hoá và tính điểm cho 20 triệu bản ghi. Phản xạ sai: ProcessPoolExecutor chia 8 lõi. Phản xạ đúng: hỏi trước "logic này vector hoá được không?". Nếu là số học/biến đổi cột → viết lại bằng Polars/NumPy, chạy trong Rust/C đa lõi, nhanh hơn multiprocessing mà không tốn chi phí pickle 20 triệu bản ghi. Chỉ khi mỗi bản ghi cần logic Python thuần không thể vector hoá (parse chuỗi bất quy tắc, gọi lib Python từng dòng) thì mới chunk dữ liệu + ProcessPoolExecutor, để mỗi tiến trình xử lý một lô lớn nhằm khấu hao chi phí khởi tạo và serialization.

Ghi nhớ

• GIL: một tiến trình CPython chỉ chạy một thread bytecode Python tại một thời điểm → threading vô ích cho CPU-bound, nhưng tốt cho I/O-bound (GIL nhả khi chờ). Thư viện C (NumPy/Polars) nhả GIL. No-GIL đã supported từ 3.14 nhưng chưa mặc định.
• Chọn công cụ theo tác vụ: I/O-bound → threading (ít việc) hoặc asyncio (quy mô lớn); CPU-bound Python thuần → multiprocessing; CPU-bound dạng bảng → vector hoá.
• Vector hoá thắng vòng lặp: đừng song song hoá for — xoá nó bằng NumPy/pandas/Polars.
• Đo trước khi tối ưu: timeit → cProfile → line_profiler → py-spy (production). Sửa đúng cái 20% nặng nhất.
• Bộ nhớ là hiệu năng: generator/lazy, xử lý theo chunk/stream, dtype gọn, memory_profiler tìm chỗ phình.
• Mẹo khác: lru_cache, batch I/O, giảm serialization, numba/Cython cho hotspot số học cuối cùng.
• Nguyên tắc lớn nhất: đừng tối ưu sớm; và rất nhiều khi đổi công cụ (Polars/DuckDB/Spark) tốt hơn vắt kiệt CPython bằng tay.

Bài kế tiếp — Production & chất lượng — bàn về việc đưa những pipeline nhanh này vào vận hành ổn định: logging, retry, test và giám sát.

Nguồn tham khảo

• PEP 703 – Making the Global Interpreter Lock Optional in CPython
• Python support for free threading — docs.python.org
• Faster Python: Unlocking the Python Global Interpreter Lock — JetBrains Blog
• Concurrency in async/await and threading — JetBrains Blog
• asyncio vs threading vs multiprocessing: When to Use Each — BSWEN
• The Python Profilers — docs.python.org
• py-spy: Sampling profiler for Python programs — GitHub
• How to Profile Python Applications with cProfile and py-spy — OneUptime