Posted inPython

Xử lý file Parquet trong Python bằng PyArrow: Đọc, ghi và kinh nghiệm tối ưu trên production

Python tutorial - IT technology blog
Python tutorial - IT technology blog

Sự cố lúc 3 giờ sáng và bài toán 12GB CSV

Hệ thống của bên mình có một cron job chạy lúc 3 giờ sáng. Nhiệm vụ của nó là tổng hợp access log và event tracking từ các cụm web server. Cách đây nửa năm, lượng log đổ về mỗi ngày cán mốc 12GB dạng CSV thô.

Hậu quả đến rất nhanh. Script chạy lệnh pd.read_csv() quen thuộc ngốn sạch 16GB RAM của VPS chỉ sau 4 phút. CPU chạm 100%. Linux kernel kích hoạt OOM Killer hạ gục tiến trình ngay lập tức.

Lúc đó mình có hai đường lui. Một là nâng cấp RAM lên 32GB hoặc 64GB, tốn thêm 40-60 USD mỗi tháng cho server. Hai là đập đi xây lại định dạng lưu trữ và luồng đọc dữ liệu. Mình chọn cách thứ hai: chuyển toàn bộ sang Apache Parquet và dùng PyArrow làm engine xử lý chính.

Parquet là định dạng lưu trữ theo cột (columnar storage), đối lập hoàn toàn với cách dàn phẳng theo từng dòng của CSV. Ba ưu điểm kỹ thuật giúp Parquet vượt trội:

  • Nén dữ liệu vượt trội: Các giá trị trong cùng một cột có cùng kiểu dữ liệu. Thuật toán Snappy hoặc ZSTD nhờ đó gom cụm và nén hiệu quả hơn hẳn text thô. File CSV 12GB nén lại chỉ còn khoảng 1.4GB đến 1.8GB. Tiết kiệm hơn 85% dung lượng ổ đĩa.
  • Projection và Predicate Pushdown: Khi query 3 cột trong bảng có 50 cột, PyArrow chỉ đọc đúng dải byte của 3 cột đó từ disk. Toàn bộ 47 cột còn lại bị bỏ qua hoàn toàn. Với CSV, parser buộc phải quét từ byte đầu tiên đến byte cuối cùng của toàn bộ file.
  • Zero-copy read: PyArrow ánh xạ trực tiếp bộ nhớ đệm (memory-mapped file) từ C++ Arrow C Data Interface sang Python runtime. Thao tác này loại bỏ chi phí serialize và cấp phát lại vùng nhớ. Tốc độ đọc nhờ đó tăng vọt.

Sau 6 tháng chạy ổn định trên production, thời gian xử lý của job giảm từ 18 phút xuống 42 giây. Mức đỉnh RAM giữ ổn định ở mức 780MB thay vì chạm trần 16GB như trước. Dưới đây là kiến trúc triển khai chi tiết.

Cài đặt môi trường

PyArrow có sẵn pre-built binary wheel trên PyPI cho Linux, macOS và Windows. Để hỗ trợ nén mức cao nhất, bạn cần cài đặt thêm thư viện zstandard:

# Cài đặt PyArrow và thư viện nén zstandard
pip install pyarrow zstandard

# Tùy chọn: cài đặt pandas để tương thích ngược khi cần export DataFrame
pip install pandas

Sau khi cài xong, kiểm tra nhanh codec nén trong terminal Python:

import pyarrow as pa
import pyarrow.parquet as pq

print(f"PyArrow version: {pa.__version__}")
print(f"ZSTD supported: {pa.Codec.is_available('zstd')}")
print(f"Snappy supported: {pa.Codec.is_available('snappy')}")

Nếu cả hai dòng kiểm tra đều trả về True, hệ thống đã sẵn sàng làm việc với các file Parquet nén sâu.

Triển khai chuẩn trên Production: Đọc, Ghi và Tối ưu

1. Ghi Parquet với Schema tường minh

Sai lầm phổ biến nhất khi viết file Parquet là để engine tự suy diễn kiểu (type inference). Việc này vừa tốn CPU, vừa dễ gây lỗi schema mismatch giữa các mẻ log khác nhau. Hãy luôn định nghĩa pa.schema chặt chẽ ngay từ đầu:

import pyarrow as pa
import pyarrow.parquet as pq
from datetime import datetime

# 1. Khai báo Schema tường minh, chọn kiểu dữ liệu nhỏ nhất có thể
schema = pa.schema([
    ('event_id', pa.string()),
    ('user_id', pa.int64()),
    ('action', pa.string()),
    ('latency_ms', pa.float32()),  # Dùng float32 thay vì float64 để giảm 50% RAM
    ('created_at', pa.timestamp('ms')),
    ('is_bot', pa.bool_())
])

# 2. Dữ liệu mảng mẫu
data = [
    pa.array(['evt_101', 'evt_102', 'evt_103']),
    pa.array([10001, 10002, 10003]),
    pa.array(['login', 'view_item', 'checkout']),
    pa.array([45.2, 120.8, 310.5], type=pa.float32()),
    pa.array([datetime.now(), datetime.now(), datetime.now()], type=pa.timestamp('ms')),
    pa.array([False, False, True])
]

# 3. Tạo bảng PyArrow Table
table = pa.Table.from_arrays(data, schema=schema)

# 4. Ghi ra đĩa với ZSTD level 3 và row group 50.000 dòng
pq.write_table(
    table,
    'events_2026_10.parquet',
    compression='zstd',
    compression_level=3,
    row_group_size=50000
)
print("Ghi file Parquet thành công!")

Tham số row_group_size mang tính sống còn. Nếu đặt quá nhỏ (dưới 5.000 dòng), metadata phình to khiến tốc độ đọc chậm đi rõ rệt. Ngược lại, nếu gom hàng triệu dòng vào một row group, bạn mất khả năng đọc song song và lọc phân vùng. Ngưỡng 50.000 đến 100.000 dòng là điểm cân bằng tối ưu cho hầu hết hệ thống log backend.

2. Đọc tối ưu: Projection và PyArrow Compute

Đừng bao giờ đọc toàn bộ file nếu chỉ cần tính toán một vài chỉ số. Dùng tham số columns để kéo đúng dữ liệu cần thiết:

import pyarrow.parquet as pq
import pyarrow.compute as pc

# Chỉ nạp 2 cột vào bộ nhớ: user_id và latency_ms
table_subset = pq.read_table(
    'events_2026_10.parquet',
    columns=['user_id', 'latency_ms']
)

print(f"Số bản ghi: {table_subset.num_rows}")
print(f"RAM sử dụng: {table_subset.nbytes / (1024 * 1024):.2f} MB")

# Tính toán trực tiếp bằng PyArrow Compute trên C++, không cần convert sang pandas
avg_latency = pc.mean(table_subset['latency_ms'])
print(f"Độ trễ trung bình: {avg_latency.as_py():.2f} ms")

Mẹo nhỏ: việc tính toán trực tiếp qua pyarrow.compute giữ nguyên cấu trúc bộ nhớ dạng C++. Bạn tránh được bước copy dữ liệu sang NumPy hoặc pandas, tiết kiệm đáng kể CPU khi chạy định kỳ.

3. Phân vùng dữ liệu (Hive Partitioning)

Khi dữ liệu tích lũy từ vài triệu lên hàng trăm triệu bản ghi, gom tất cả vào một file duy nhất sẽ gây tắc nghẽn I/O. Hãy tổ chức dữ liệu theo dạng cây thư mục action=login/ hoặc date=2026-10-27/:

import pyarrow.dataset as ds
import pyarrow.parquet as pq

# Ghi dataset tự động phân thư mục theo cột action
pq.write_to_dataset(
    table,
    root_path='events_partitioned',
    partition_cols=['action'],
    compression='zstd'
)

# Đọc dữ liệu qua Dataset API với Hive partitioning
dataset = ds.dataset('events_partitioned', format='parquet', partitioning='hive')

# Lọc dữ liệu: PyArrow chỉ mở đúng folder action=checkout, bỏ qua các folder khác
checkout_filter = ds.field('action') == 'checkout'
filtered_table = dataset.to_table(filter=checkout_filter)

print(f"Số event checkout quét được: {filtered_table.num_rows}")

Kiểm tra hiệu năng và Monitoring

Đừng đoán mò hiệu năng. Bạn có thể đo lường trực tiếp dung lượng đĩa và thời gian bóc tách metadata bằng đoạn mã dưới đây:

import os
import time
import pyarrow.parquet as pq

file_path = 'events_2026_10.parquet'

# 1. Kích thước file vật lý trên ổ cứng
size_mb = os.path.getsize(file_path) / (1024 * 1024)
print(f"[Disk Usage] Kích thước file: {size_mb:.2f} MB")

# 2. Đọc metadata mà không cần nạp toàn bộ body data
parquet_file = pq.ParquetFile(file_path)
meta = parquet_file.metadata
print(f"[Metadata] Tổng số dòng: {meta.num_rows:,}")
print(f"[Metadata] Số lượng Row Groups: {meta.num_row_groups}")
print(f"[Metadata] Thuật toán nén: {meta.row_group(0).column(0).compression}")

# 3. Benchmark thời gian đọc 1 cột cụ thể
t0 = time.perf_counter()
col_data = parquet_file.read(columns=['user_id'])
elapsed = (time.perf_counter() - t0) * 1000

print(f"[Benchmark] Thời gian đọc cột 'user_id': {elapsed:.2f} ms")
print(f"[Memory] RAM nạp cột: {col_data.nbytes / (1024 * 1024):.4f} MB")

Khi đưa vào production, bạn cần giám sát 3 chỉ số sau:

  1. Compression Ratio: So sánh dung lượng trong RAM (table.nbytes) với dung lượng file trên đĩa. Log hệ thống thông thường cần đạt tỷ lệ nén từ 1:4 đến 1:7 với ZSTD level 3.
  2. Small Files Problem: Một file 50MB nhưng có hơn 500 Row Groups là dấu hiệu xấu. Chi phí duyệt header metadata sẽ ngốn sạch lợi thế của Parquet. Khi gặp tình trạng này, hãy gộp (compact) các batch nhỏ lại trước khi commit ra đĩa.
  3. Peak Memory (RSS): Chạy script qua lệnh /usr/bin/time -v python script.py trên Linux. Kiểm tra dòng Maximum resident set size. Thông số này phải duy trì ổn định qua từng ngày thay vì tăng tuyến tính theo kích thước toàn bộ kho dữ liệu.

Chuyển từ CSV sang Parquet kết hợp PyArrow giải quyết tận gốc bài toán nghẽn cổ chai I/O. Với các tác vụ dữ liệu vừa và lớn, đây là giải pháp tối ưu tài nguyên phần cứng hiệu quả trước khi nghĩ đến việc bỏ tiền nâng cấp server.

Share:
Tags: