Python DE 2 — pandas chuyên sâu & hiệu năng

Vì sao cần một bài "chuyên sâu" về pandas?

Ở bài Tổng quan chúng ta đã dựng khung tư duy Data Engineering với Python. Bài này đi sâu vào công cụ được dùng nhiều nhất trong đời sống hàng ngày của một DE: pandas. Vấn đề là phần lớn người dùng học pandas theo kiểu "chạy được là xong" — và rồi khi dữ liệu tăng từ vài nghìn dòng lên vài chục triệu dòng, code chậm hàng chục phút, hoặc tệ hơn là MemoryError.

Bài này tập trung vào pandas 2.x với ba câu hỏi cốt lõi: (1) làm sao viết pandas nhanh (vectorization), (2) làm sao dùng ít RAM (dtype & Arrow), và (3) làm sao xử lý dữ liệu lớn hơn RAM (chunk). Toàn bộ ví dụ là minh hoạ — sandbox của khoá này là PostgreSQL, Python không chạy trực tiếp ở đây.

1. Cấu trúc dữ liệu: Series, DataFrame và cái bẫy index

Series là một cột có nhãn (index) + một mảng giá trị đồng nhất kiểu (dựa trên NumPy hoặc Arrow). DataFrame là tập hợp các Series dùng chung một index dòng. Hiểu đúng index là chìa khoá dùng pandas hiệu quả:

• Index căn chỉnh tự động (alignment): khi cộng hai Series, pandas ghép theo nhãn chứ không theo vị trí. Đây là tính năng mạnh nhưng cũng là bẫy — cộng hai Series có index lệch nhau ra NaN.
• Index dùng để tra cứu (.loc) nhanh nếu nó được sắp xếp; tra cứu trên index chưa sắp xếp có thể chậm.
• Sau groupby, set_index, merge... index dễ bị "rối". Khi cần đánh số lại dùng reset_index(drop=True).

# (minh hoạ) alignment theo nhãn, không theo vị trí
import pandas as pd
a = pd.Series([1, 2, 3], index=["x", "y", "z"])
b = pd.Series([10, 20, 30], index=["y", "z", "w"])
a + b        # x=NaN, y=12, z=23, w=NaN  -> alignment theo index

Arrow-backed dtypes (pandas 2.x)

pandas 2.0 mở ra khả năng lưu dữ liệu bằng backend Apache Arrow thay cho NumPy. Với dữ liệu thật của DE, hai lợi ích lớn nhất:

• Chuỗi (string): cột string kiểu Arrow tốn ít hơn tới ~70% RAM so với cột object (NumPy). Đây là con số rất đáng kể vì dữ liệu ETL đầy chuỗi (mã KH, tên chi nhánh, loại giao dịch).
• NA thống nhất: Arrow có giá trị "missing" thực sự cho cả số nguyên — không còn cảnh cột int bị ép thành float chỉ vì có một ô trống.

# (minh hoạ) đọc với backend Arrow
df = pd.read_parquet("giao_dich.parquet", dtype_backend="pyarrow")
df = pd.read_csv("gd.csv", dtype_backend="pyarrow")
s = pd.Series(["A", "B", None], dtype="string[pyarrow]")   # chỉ bật Arrow cho string

Lưu ý thực dụng: backend Arrow tổng quát vẫn đang trưởng thành. Khuyến nghị an toàn là bật Arrow cho string trước (thành phần chín nhất), còn phần còn lại chuyển dần khi workflow đã kiểm thử ổn định.

2. Vectorization — nền tảng của pandas nhanh

Đây là điều quan trọng nhất trong cả bài. Vectorization nghĩa là thực hiện phép tính trên cả cột một lần, để công việc lặp diễn ra trong C/NumPy/Arrow (biên dịch, chạy trên bộ nhớ liền khối) thay vì trong vòng lặp Python (mỗi vòng phải diễn giải bytecode, đóng gói/mở gói object).

Chênh lệch tốc độ không nhỏ: các benchmark phổ biến cho thấy vectorization nhanh hơn iterrows() cỡ hàng trăm lần, và nhanh hơn apply() khoảng hàng chục lần (trên tập ~10k dòng, iterrows ~0.74s, apply ~0.13s, vectorized ~0.001s). Trên dữ liệu triệu dòng, khác biệt là "vài phút" so với "dưới một giây".

# (minh hoạ) TÍNH PHÍ = 0.1% số tiền, tối thiểu 1000

# CÁCH TỆ NHẤT: vòng lặp iterrows -> chậm nhất
fees = []
for _, row in df.iterrows():           # mỗi vòng tạo 1 Series -> rất tốn
    fees.append(max(row["amount"] * 0.001, 1000))
df["fee"] = fees

# TỐT HƠN: apply theo dòng (vẫn gọi Python cho từng dòng)
df["fee"] = df["amount"].apply(lambda x: max(x * 0.001, 1000))

# TỐT NHẤT: vectorize hoàn toàn
df["fee"] = (df["amount"] * 0.001).clip(lower=1000)

Điều kiện: dùng np.where / .loc thay if-else từng dòng

# (minh hoạ) gắn nhãn giao dịch lớn
import numpy as np

# vectorized: np.where cho điều kiện đơn
df["flag"] = np.where(df["amount"] >= 500_000_000, "LON", "THUONG")

# nhiều nhánh: np.select
conds  = [df["amount"] >= 1e9, df["amount"] >= 5e8]
labels = ["RAT_LON", "LON"]
df["muc"] = np.select(conds, labels, default="THUONG")

# gán có điều kiện bằng .loc (boolean mask)
df.loc[df["type"] == "RUT_TIEN", "amount"] *= -1

Khi nào buộc phải dùng apply / vòng lặp

Vectorization không phải lúc nào cũng khả thi. Chấp nhận apply (hoặc itertuples, nhanh hơn iterrows nhiều vì trả về namedtuple thay vì Series) khi:

• Logic quá phức tạp không diễn đạt được bằng phép trên cột (ví dụ parse chuỗi bất quy tắc, gọi hàm bên thứ ba).
• Nút cổ chai nằm ở I/O bên ngoài (gọi API làm giàu dữ liệu từng dòng): lúc này thời gian mạng áp đảo, chênh lệch iterrows/itertuples không còn quan trọng, nên chọn cách dễ đọc nhất.

Nhưng nguyên tắc: luôn thử vectorize trước, chỉ rơi xuống apply khi thực sự cần.

3. dtypes & bộ nhớ — nửa còn lại của hiệu năng

Chậm và tốn RAM thường có chung một gốc: dtype mặc định quá "béo". pandas mặc định dùng int64/float64 và object cho chuỗi — trong khi phần lớn dữ liệu không cần.

Bước đầu tiên là đo:

# (minh hoạ) đo bộ nhớ thật (deep=True mới tính đúng chuỗi object)
df.info(memory_usage="deep")
df.memory_usage(deep=True).sort_values(ascending=False)

Ba đòn bẩy chính:

(a) Downcast số. int64 -> int32/int16, float64 -> float32 nếu dải giá trị cho phép.

# (minh hoạ)
df["so_luong"] = pd.to_numeric(df["so_luong"], downcast="integer")   # int8/16/32
df["ty_gia"]   = pd.to_numeric(df["ty_gia"], downcast="float")       # float32

(b) category cho chuỗi lặp lại nhiều (low-cardinality). Cột như loai_gd, chi_nhanh, ma_tien_te chỉ có vài chục giá trị khác nhau trên hàng triệu dòng — lưu dạng category (mã hoá số + bảng tra) giảm RAM cực mạnh và tăng tốc groupby.

# (minh hoạ)
df["chi_nhanh"] = df["chi_nhanh"].astype("category")
df["loai_gd"]   = df["loai_gd"].astype("category")

(c) Chuyển string sang Arrow ("string[pyarrow]") cho cột chuỗi tự do (cardinality cao) — giảm RAM đáng kể so với object.

(d) Chỉ đọc cột cần ngay từ khâu I/O — đừng nạp cả bảng rồi mới bỏ cột:

# (minh hoạ) đọc gọn ngay từ đầu
df = pd.read_csv(
    "giao_dich.csv",
    usecols=["ma_kh", "amount", "loai_gd", "ngay"],   # chỉ 4 cột
    dtype={"loai_gd": "category", "ma_kh": "string[pyarrow]"},
    parse_dates=["ngay"],
)

Kết hợp usecols + dtype + định dạng nhị phân (Parquet) có thể giảm thời gian nạp và RAM đi 10–100 lần so với đọc CSV thô. Xem thêm ở bài File & I/O.

4. groupby / agg — gộp nhóm đúng cách

groupby là "SQL GROUP BY" của pandas. Ba họ hàm cần phân biệt rõ:

• agg: mỗi nhóm -> một giá trị (thu gọn). Dùng khi cần bảng tổng hợp.
• transform: mỗi nhóm -> trả về cùng số dòng (broadcast kết quả về từng dòng). Dùng để tạo cột "tỷ trọng trong nhóm", "trung bình nhóm gắn vào dòng".
• apply: linh hoạt nhất nhưng chậm nhất — dùng khi hai cái trên không đủ.

# (minh hoạ) named aggregation: cột kết quả rõ ràng
bao_cao = (
    df.groupby("chi_nhanh", observed=True)   # observed=True: bỏ nhóm rỗng của category
      .agg(
          tong_tien=("amount", "sum"),
          so_gd=("amount", "size"),
          tien_tb=("amount", "mean"),
          tien_max=("amount", "max"),
      )
      .reset_index()
)

# transform: gắn trung bình nhóm về từng dòng, rồi tính lệch
df["tb_chi_nhanh"] = df.groupby("chi_nhanh")["amount"].transform("mean")
df["lech"] = df["amount"] - df["tb_chi_nhanh"]

Với category dtype, luôn cân nhắc observed=True để không sinh ra tổ hợp nhóm rỗng.

5. merge / join — nơi dữ liệu hay "nổ dòng"

merge là JOIN. Lỗi kinh điển của DE: sau khi join, số dòng phình lên gấp nhiều lần vì quan hệ không phải 1-1 như tưởng (khoá bên phải bị trùng), tạo ra tích Descartes cục bộ.

Hai vũ khí phòng thủ: validate và indicator.

# (minh hoạ) join an toàn
kq = df.merge(
    dim_kh,                       # bảng khách hàng
    on="ma_kh",
    how="left",
    validate="many_to_one",       # NỔ nếu ma_kh trong dim_kh bị trùng -> báo lỗi ngay
    indicator=True,               # thêm cột _merge: left_only / both / right_only
)

# kiểm tra dòng không khớp
kq["_merge"].value_counts()
kq[kq["_merge"] == "left_only"]   # KH không tìm thấy trong bảng dim

• validate nhận "one_to_one", "one_to_many", "many_to_one", "many_to_many" — pandas sẽ raise nếu quan hệ thực tế vi phạm giả định. Luôn khai báo khi bạn nghĩ mình biết quan hệ.
• indicator=True giúp audit chất lượng join sau ETL.
• Đảm bảo dtype của khoá join khớp nhau (một bên int, một bên string sẽ không khớp và cho toàn NaN).

concat dùng để nối theo dọc (thêm dòng) hoặc ngang (thêm cột); nhớ ignore_index=True khi nối dòng để tránh index trùng.

6. pivot / reshape — xoay bảng

# (minh hoạ) pivot_table: bảng chéo chi nhánh x loại giao dịch
bang = pd.pivot_table(
    df, index="chi_nhanh", columns="loai_gd",
    values="amount", aggfunc="sum", fill_value=0,
)

# melt: wide -> long (chuẩn hoá cho lưu kho/ghi DB)
dai = bang.reset_index().melt(
    id_vars="chi_nhanh", var_name="loai_gd", value_name="tong_tien"
)

# stack / unstack: xoay giữa cột và mức index
df.groupby(["chi_nhanh", "loai_gd"])["amount"].sum().unstack(fill_value=0)

Quy tắc: dữ liệu long (mỗi dòng một quan sát) hợp để lưu trữ và ghi DB; dữ liệu wide hợp để đọc/báo cáo. ETL thường chuẩn hoá về long.

7. Thời gian — thứ luôn có trong ETL

# (minh hoạ)
df["ngay"] = pd.to_datetime(df["ngay"], format="%Y-%m-%d", errors="coerce")
# format tường minh -> nhanh & tránh đoán sai; errors="coerce" -> giá trị hỏng thành NaT

# resample: tổng giao dịch theo ngày (index phải là datetime)
theo_ngay = df.set_index("ngay")["amount"].resample("D").sum()

# rolling: trung bình động 7 ngày
theo_ngay.rolling(7).mean()

# timezone: chuẩn hoá về UTC khi gộp dữ liệu nhiều nguồn
ts = df["thoi_gian"].dt.tz_localize("Asia/Ho_Chi_Minh").dt.tz_convert("UTC")

# truy xuất thành phần ngày bằng accessor .dt (vectorized)
df["nam"]  = df["ngay"].dt.year
df["thang"] = df["ngay"].dt.month

Mẹo ETL: luôn truyền format= cho to_datetime khi biết định dạng (nhanh hơn nhiều và tránh parse nhầm ngày/tháng); dùng errors="coerce" để cô lập dòng hỏng thay vì làm sập cả pipeline.

8. Missing / duplicate

# (minh hoạ)
df["amount"] = df["amount"].fillna(0)
df = df.dropna(subset=["ma_kh"])            # bắt buộc phải có mã KH

# trùng lặp
df.duplicated(subset=["ma_gd"]).sum()       # đếm giao dịch trùng ID
df = df.drop_duplicates(subset=["ma_gd"], keep="last")

Trong ETL, drop_duplicates(subset=..., keep=...) là bước khử trùng khi gộp dữ liệu tải lại (re-run) — chọn keep="last" nếu bản ghi mới ghi đè bản cũ.

9. method chaining & pipe — code sạch, ít biến trung gian

# (minh hoạ) một luồng đọc-được, không tạo df tạm rải rác
ket_qua = (
    df
    .loc[df["amount"] > 0]
    .assign(fee=lambda d: (d["amount"] * 0.001).clip(lower=1000))
    .groupby("chi_nhanh", observed=True)
    .agg(tong=("amount", "sum"), phi=("fee", "sum"))
    .reset_index()
    .pipe(lambda d: d[d["tong"] > 1e9])     # pipe: cắm hàm tuỳ ý vào chuỗi
)

SettingWithCopyWarning: copy vs view

Cảnh báo này xuất hiện khi bạn gán vào một lát cắt (slice) mà pandas không chắc là copy hay view của DataFrame gốc — kết quả gán có thể "trôi" đi đâu không rõ.

# (minh hoạ) SAI: gán chuỗi trên slice
tmp = df[df["amount"] > 0]
tmp["fee"] = 0            # cảnh báo: tmp có thể là view

# ĐÚNG: .loc một bước, hoặc .copy() tường minh
df.loc[df["amount"] > 0, "fee"] = 0
tmp = df[df["amount"] > 0].copy()
tmp["fee"] = 0            # yên tâm: tmp là bản độc lập

Tin tốt: pandas đang tiến tới cơ chế Copy-on-Write (mặc định từ pandas 3.0), làm ngữ nghĩa copy/view nhất quán và loại bỏ phần lớn cảnh báo này. Dù vậy, thói quen dùng .loc một bước và .copy() tường minh vẫn là cách viết an toàn nhất.

10. Dữ liệu lớn hơn RAM — đọc theo chunk

pandas giữ toàn bộ DataFrame trong RAM. Một file 20GB không vào nổi máy 16GB. Giải pháp trong khuôn khổ pandas: đọc theo chunksize, xử lý từng mảnh, gộp kết quả (đã thu nhỏ) lại.

# (minh hoạ) tổng tiền theo chi nhánh trên file khổng lồ, RAM thấp
tong = None
reader = pd.read_csv(
    "giao_dich_lon.csv",
    usecols=["chi_nhanh", "amount"],       # lọc CỘT sớm
    dtype={"chi_nhanh": "category"},
    chunksize=500_000,                     # đọc 500k dòng mỗi lần
)
for chunk in reader:
    chunk = chunk[chunk["amount"] > 0]     # lọc DÒNG sớm
    phan = chunk.groupby("chi_nhanh", observed=True)["amount"].sum()
    tong = phan if tong is None else tong.add(phan, fill_value=0)

tong = tong.sort_values(ascending=False)

Nguyên tắc vàng khi xử lý theo chunk: lọc cột và dòng càng sớm càng tốt, chỉ giữ lại phần đã thu nhỏ (kết quả tổng hợp) — đừng tích luỹ toàn bộ chunk vào một list rồi concat (như thế lại quay về nạp hết vào RAM).

Khi bài toán bắt đầu "khó thở" với pandas dù đã tối ưu, đó là tín hiệu chuyển công cụ: polars (đa luồng, lazy, out-of-core) hoặc duckdb (query SQL thẳng trên Parquet/CSV, xử lý dữ liệu lớn hơn RAM). Chi tiết ở bài Polars & Arrow.

Use case thực tế: xử lý file giao dịch 15GB

Bối cảnh: file giao_dich_thang.csv ~15GB (80 triệu dòng), 40 cột, máy 16GB RAM. Yêu cầu: tính tổng tiền và số giao dịch theo chi_nhanh x loai_gd, chỉ lấy giao dịch dương, ghi ra Parquet.

Chiến lược gộp cả ba đòn bẩy: giảm RAM (dtype) + vectorize + chunk.

# (minh hoạ) pipeline out-of-core
import pandas as pd

CHUNK = 1_000_000
acc = []   # chỉ tích luỹ kết quả ĐÃ TỔNG HỢP (nhỏ), không phải dữ liệu thô

reader = pd.read_csv(
    "giao_dich_thang.csv",
    usecols=["chi_nhanh", "loai_gd", "amount", "phi"],   # 4/40 cột
    dtype={
        "chi_nhanh": "category",
        "loai_gd":   "category",
    },
    chunksize=CHUNK,
)

for chunk in reader:
    chunk = chunk.loc[chunk["amount"] > 0]                 # lọc dòng sớm
    chunk["amount"] = pd.to_numeric(chunk["amount"], downcast="float")
    part = (
        chunk.groupby(["chi_nhanh", "loai_gd"], observed=True)
             .agg(tong=("amount", "sum"), so_gd=("amount", "size"))
    )
    acc.append(part)

# gộp các tổng hợp cục bộ -> tổng hợp lần cuối
ket_qua = (
    pd.concat(acc)
      .groupby(level=["chi_nhanh", "loai_gd"], observed=True)
      .sum()
      .reset_index()
)
ket_qua.to_parquet("bao_cao_thang.parquet", index=False)

Điểm mấu chốt: RAM tại mỗi thời điểm chỉ chứa 1 chunk (1 triệu dòng x 4 cột đã ép dtype), phần tích luỹ acc là các bảng tổng hợp bé xíu. Không bao giờ có 80 triệu dòng cùng lúc trong bộ nhớ. Với category cho hai cột khoá và downcast cho amount, mỗi chunk nhẹ hơn nhiều lần so với đọc thô.

Ghi nhớ

• Vectorize trước, apply sau, vòng lặp cuối cùng. Phép trên cột / np.where / np.select / accessor .str .dt nhanh hơn apply hàng chục lần và hơn iterrows hàng trăm lần.
• Đo rồi mới tối ưu RAM: df.info(memory_usage="deep"). Downcast số, category cho chuỗi lặp, string Arrow cho chuỗi tự do.
• Đọc gọn từ đầu: usecols + dtype + parse_dates; ưu tiên Parquet hơn CSV.
• merge an toàn: luôn khai báo validate=, dùng indicator=True để audit, khớp dtype khoá join — tránh nổ dòng.
• Phân biệt agg / transform / apply; dùng observed=True với category.
• to_datetime truyền format=, errors="coerce"; resample/rolling cần index datetime.
• Chunk cho dữ liệu lớn hơn RAM: lọc cột & dòng sớm, chỉ giữ kết quả tổng hợp. Khi pandas hụt hơi, chuyển sang polars/duckdb.
• Viết method chaining; né SettingWithCopyWarning bằng .loc một bước và .copy() tường minh (Copy-on-Write sẽ là mặc định ở pandas 3.0).

Đọc tiếp: Polars & Arrow, File & I/O, Hiệu năng & song song.