Posted inProgramming

Tối Ưu Xử Lý Dữ Liệu Lớn trong Python với Polars: LazyFrame, Streaming và Zero-Copy

Programming tutorial - IT technology blog
Programming tutorial - IT technology blog

Khi Pandas Bắt Đầu Cướp Giấc Ngủ Của Bạn

Năm ngoái, tôi được giao một pipeline phân tích 15 GB log sự kiện mỗi đêm. Script chạy ổn định trong nhiều tháng — cho đến khi lượng dữ liệu tăng gấp đôi. Đột nhiên chúng tôi gặp lỗi OOM-kill trên máy 32 GB, phải khởi động lại lúc 2 giờ sáng, rồi dùng đến giải pháp vá víu kinh điển là chia nhỏ read_csv — biến công việc 20 phút thành mớ hỗn độn 90 phút.

Pandas rất xuất sắc cho công việc khám phá dữ liệu và dataset nhỏ đến vừa, nhưng nó chưa bao giờ được thiết kế cho điều này. Nó tải toàn bộ dữ liệu vào bộ nhớ ngay lập tức, sao chép dữ liệu giữa các phép toán, và chỉ dùng một CPU core mặc định. Thiết kế đó hợp lý vào năm 2008. Nhưng không còn hợp lý khi bạn xử lý hàng trăm triệu hàng.

Polars đã thay đổi hoàn toàn cách tôi nghĩ về xử lý dữ liệu. Sau khi chuyển đổi pipeline đó, bộ nhớ đỉnh giảm 70%, thời gian chạy từ 90 phút xuống còn chưa đầy 8 phút, và tôi không còn nhận cảnh báo lúc 2 giờ sáng nữa. Đây không phải phép màu — đây là mô hình thực thi khác biệt, và một khi bạn hiểu được nó, bạn sẽ chọn Polars trước tiên cho mọi tác vụ vượt qua mức dataset thử nghiệm.

Những Ý Tưởng Cốt Lõi Làm Polars Khác Biệt

Thực Thi Lười và Lập Kế Hoạch Query

Khi bạn viết một LazyFrame trong Polars, thực tế không có gì chạy cả. Bạn đang xây dựng một query plan — một mô tả về những gì bạn muốn. Polars thu thập toàn bộ plan, tối ưu hóa nó (predicate pushdown, projection pushdown, loại bỏ CSE), và chỉ thực thi khi bạn gọi .collect().

So sánh với Pandas, nơi mỗi phương thức được gọi nối tiếp đều tạo ra một DataFrame trung gian đầy đủ. Với Polars LazyFrame, nếu bạn lọc xuống còn 5% số hàng ngay từ đầu plan, mọi bước tiếp theo chỉ làm việc trên 5% dữ liệu — không phải 100%.

Streaming Engine cho Xử Lý Ngoài Bộ Nhớ

Streaming engine là thứ giúp xử lý được các file thực sự lớn. Thay vì tải toàn bộ dataset vào bộ nhớ cùng lúc, Polars xử lý theo từng chunk trong nội bộ — nhưng khác với chunking trong Pandas, bạn không cần viết logic chia chunk. Chỉ cần thêm streaming=True vào .collect() và Polars tự lo phần còn lại. Query plan được giữ nguyên; Polars tự tìm cách chia công việc qua các chunk.

Zero-Copy và Apache Arrow Phía Dưới

Polars được xây dựng trên Apache Arrow làm định dạng bộ nhớ. Arrow là định dạng dữ liệu dạng cột, zero-copy. Khi bạn cắt một cột hoặc truyền dữ liệu giữa các thư viện hỗ trợ Arrow — DuckDB, PyArrow, Pandas 2.x với dtype_backend="pyarrow" — không có dữ liệu nào bị sao chép. Bạn chuyển một con trỏ, không phải một bản sao.

Nghe có vẻ trừu tượng cho đến khi bạn theo dõi nó trong profiler. Tôi lần đầu nhận ra sự khác biệt khi kết hợp pipeline DuckDB→Polars→scikit-learn: zero byte được sao chép qua các lần chuyển giao đó. Đây là chi phí ẩn lớn nhất trong các workflow đa thư viện, và Arrow loại bỏ hoàn toàn điều đó.

SIMD và Đa Nhân Mặc Định

Polars được viết bằng Rust và dùng Rayon để xử lý song song dựa trên thread. Các phép tính tổng hợp, join và biến đổi tự động phân tán trên tất cả các CPU core hiện có. Bạn không cần bật tính năng này — bạn tắt nó đi nếu cần.

Thực Hành: Xây Lại Pipeline Thực Tế

Cài Đặt

pip install polars pyarrow

LazyFrame Cơ Bản

Bắt đầu bằng cách scan một file CSV lớn mà không cần tải nó:

import polars as pl

# scan_csv chỉ đọc schema — chưa có dữ liệu nào được tải
lf = pl.scan_csv("events.csv")

# Xây dựng query plan
result = (
    lf
    .filter(pl.col("status") == "error")
    .filter(pl.col("response_ms") > 500)
    .select(["timestamp", "endpoint", "user_id", "response_ms"])
    .group_by("endpoint")
    .agg(
        pl.col("user_id").count().alias("error_count"),
        pl.mean("response_ms").alias("avg_response_ms"),
        pl.max("response_ms").alias("p100_response_ms"),
    )
    .sort("error_count", descending=True)
)

# Chưa có gì chạy cả. Thực thi ngay bây giờ:
df = result.collect()
print(df.head(10))

Các lệnh filter được đẩy xuống tầng scan — Polars bỏ qua các hàng không khớp trước khi chúng chạm đến bộ nhớ. Lệnh select đảm bảo chỉ 4 cột được đọc từ đĩa thay vì toàn bộ file. Đó là predicate pushdown và projection pushdown đang hoạt động miễn phí.

Streaming Engine cho File Không Vừa trong RAM

Có file Parquet 50 GB trên máy 16 GB? Chế độ streaming xử lý được:

import polars as pl

lf = pl.scan_parquet("huge_dataset/*.parquet")  # glob cũng hoạt động

daily_summary = (
    lf
    .filter(pl.col("event_type").is_in(["purchase", "refund"]))
    .with_columns(
        pl.col("created_at").str.to_datetime().dt.date().alias("date")
    )
    .group_by(["date", "event_type"])
    .agg(
        pl.sum("amount").alias("total_amount"),
        pl.len().alias("event_count"),
    )
)

# streaming=True: Polars xử lý theo chunk, giữ bộ nhớ ổn định
df = daily_summary.collect(streaming=True)
df.write_parquet("daily_summary.parquet")

Một hạn chế cần biết: chưa phải tất cả phép toán đều có thể stream được. Một số join phức tạp yêu cầu một lượt tải đầy đủ vào bộ nhớ. Polars sẽ chuyển sang in-memory cho những bước đó và phát ra cảnh báo. Chạy pl.Config.set_verbose(True) trong quá trình phát triển để thấy chính xác nơi streaming được kích hoạt.

Chuyển Giao Zero-Copy sang Công Cụ Khác

Việc chuyển giao dữ liệu giữa các thư viện là nơi Arrow phát huy giá trị. Đây là mẫu cho các thư viện chia sẻ định dạng Arrow:

import polars as pl
import pyarrow as pa
import pandas as pd

# Polars → PyArrow: không sao chép dữ liệu
polars_df = pl.read_parquet("data.parquet")
arrow_table = polars_df.to_arrow()  # không sao chép dữ liệu

# PyArrow → Polars: không sao chép dữ liệu
polars_back = pl.from_arrow(arrow_table)

# Polars → Pandas (với Arrow backend, Pandas 2.x+): sao chép tối thiểu
pandas_df = polars_df.to_pandas(use_pyarrow_extension_array=True)

# Nếu phải đi qua numpy (Pandas cũ hơn): một lần sao chép xảy ra ở đây
pandas_legacy = polars_df.to_pandas()  # sao chép qua numpy, không tránh được

Khi cần chuyển sang thư viện chỉ hiểu Pandas (một số sklearn estimator, matplotlib), tôi giữ pipeline Polars đến tận phút chót và chỉ chuyển đổi phần dữ liệu cần thiết:

# Đừng chuyển đổi toàn bộ dataset 10 triệu hàng
# Chỉ chuyển đổi kết quả đã tổng hợp (khoảng 1000 hàng)
small_result = (
    pl.scan_csv("10M_rows.csv")
    .filter(pl.col("region") == "JP")
    .group_by("product_id")
    .agg(pl.sum("revenue"))
    .collect()
)

# Bây giờ chuyển kết quả nhỏ sang Pandas để vẽ đồ thị
small_result.to_pandas().plot.bar(x="product_id", y="revenue")

Expression API: Viết Transformation Tái Sử Dụng

Các expression trong Polars là đối tượng Python bậc nhất — điều mà Pandas chưa bao giờ làm được một cách gọn gàng. Bạn xây dựng logic biến đổi thành các biến tái sử dụng được:

import polars as pl

# Định nghĩa các expression tái sử dụng
revenue_per_visit = (pl.col("revenue") / pl.col("visits")).alias("revenue_per_visit")
bounce_rate = (pl.col("single_page_sessions") / pl.col("sessions")).alias("bounce_rate")
high_value = pl.col("revenue_per_visit") > 50

# Áp dụng chúng ở bất kỳ đâu
df = (
    pl.scan_csv("analytics.csv")
    .with_columns([revenue_per_visit, bounce_rate])
    .filter(high_value)
    .collect()
)

Đây là các đối tượng Expr thuần túy. Lưu chúng trong một list, lặp qua chúng, truyền chúng giữa các hàm — không cần monkey-patch DataFrame, không có tham chiếu cột dạng chuỗi rải rác khắp codebase.

Đo Hiệu Năng So Với Pandas

So sánh trực tiếp trên CSV 5 triệu hàng:

import time
import polars as pl
import pandas as pd

# Pandas
start = time.perf_counter()
df_pd = pd.read_csv("5M_rows.csv")
result_pd = (
    df_pd[df_pd["status"] == "active"]
    .groupby("region")["revenue"]
    .agg(["sum", "mean", "count"])
)
print(f"Pandas: {time.perf_counter() - start:.2f}s")

# Polars
start = time.perf_counter()
result_pl = (
    pl.scan_csv("5M_rows.csv")
    .filter(pl.col("status") == "active")
    .group_by("region")
    .agg([
        pl.sum("revenue"),
        pl.mean("revenue"),
        pl.len(),
    ])
    .collect()
)
print(f"Polars: {time.perf_counter() - start:.2f}s")

Trên máy của tôi (8 nhân, 16 GB RAM), Pandas mất ~4,2s, Polars mất ~0,7s — tăng tốc 6 lần trên dataset vẫn vừa vặn trong RAM. Đẩy lên 50 triệu hàng thì khoảng cách càng tệ hơn: Pandas vật lộn vượt qua 2 phút; Polars thường ở dưới 5 giây.

Khi Nào Nên Giữ Pandas (và Khi Nào Nên Chuyển)

Polars không phải là sự thay thế hoàn toàn cho mọi trường hợp sử dụng Pandas. Những đánh đổi đáng biết:

  • Giữ Pandas cho: tích hợp thư viện ML (scikit-learn, statsmodels), thư viện tài chính như QuantLib, EDA tương tác nhanh trong Jupyter khi bạn thực hiện nhiều thao tác dựa trên index theo kiểu ad-hoc.
  • Chuyển sang Polars cho: ETL pipeline, xử lý log, bất kỳ file nào trên ~1 GB, biến đổi dữ liệu production, pipeline đang tái cấu trúc để tối ưu hiệu năng, bất cứ thứ gì chạy theo lịch trình.
  • Tốt nhất từ cả hai: Dùng Polars cho công việc nặng, chuyển đổi bằng to_pandas() ở bước cuối cùng cho ML/visualization. Việc chuyển đổi dựa trên Arrow đủ nhanh để cách tiếp cận hybrid này hoạt động tốt trong thực tế.

Điều Gì Thay Đổi Sau Khi Chuyển Đổi

Quay lại pipeline chạy ban đêm đó: việc viết lại mất khoảng hai ngày, chủ yếu là học Polars expression API và debug một vài edge case của streaming. Kết quả thu được ngay lập tức — bộ nhớ đỉnh từ 28 GB xuống còn 8 GB, thời gian chạy từ 90 phút xuống còn 7 phút, không có sự cố OOM nào trong những tháng kể từ đó.

Thay đổi tư duy quan trọng nhất là chấp nhận lazy evaluation là mặc định chứ không phải ngoại lệ. Viết toàn bộ transformation của bạn, để Polars lập kế hoạch, rồi collect. Kỷ luật đó — xây dựng toàn bộ query trước khi thực thi bất kỳ phần nào — là thứ cho phép optimizer làm được việc của nó. Khi điều đó thấm vào đầu, API sẽ không còn chống lại bạn nữa.

Nếu bạn đang chạy bất kỳ Python data pipeline nào trong production hôm nay mà chưa thử Polars, nó xứng đáng để bỏ ra một buổi chiều. Bắt đầu với một script chậm, viết lại bằng scan_csv và LazyFrame, rồi đo lường. Các con số sẽ tự nói lên điều đó.

Share:
Tags: