Posted inProgramming

Phân tích dữ liệu cục bộ với DuckDB và Python: Truy vấn hàng triệu dòng mà không cần Database Server

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

Vấn đề thực tế: Khi Pandas gục ngã trước 50 triệu dòng dữ liệu

Tháng trước, đội ngũ của chúng tôi gặp phải một nút thắt cổ chai quen thuộc. Chúng tôi cần phân tích 35 GB log giao dịch hàng tháng chưa qua xử lý nằm rải rác trong 48 file CSV và Parquet. Mục tiêu rất đơn giản: tổng hợp các chỉ số thống kê, phân nhóm người dùng thành các cohort và gắn cờ các giao dịch thẻ tín dụng bất thường.

Theo thói quen, chúng tôi chọn Pandas đầu tiên. Đó là một sai lầm. Việc chạy pd.read_csv() trên một file 5.2 GB duy nhất đã đẩy dung lượng RAM thực tế vượt mức 16 GB chỉ trong vòng 40 giây. Máy trạm bị đơ, swap disk quá tải liên tục và tiến trình xử lý out-of-memory (OOM) killer của Linux đã buộc dừng tiến trình Python ngay giữa lúc đang đọc file. Chúng tôi đã thử chia nhỏ dữ liệu đầu vào bằng chunksize=100_000. Dù cách này tránh được lỗi crash, nhưng việc tính toán các phép tổng hợp lưu trạng thái (stateful aggregations) như đếm số lượng người dùng duy nhất giữa các chunk đòi hỏi hàng trăm dòng mã glue code rườm rà và dễ gãy.

Khởi chạy một container Postgres hoặc ClickHouse cục bộ bằng Docker là phương án dự phòng tiếp theo. Tuy nhiên, việc thiết lập một cơ sở dữ liệu phân tích đem lại không ít phiền toái. Bạn phải viết schema DDL cứng nhắc, tinh chỉnh shared buffers, ánh xạ cổng host và đợi 20 phút chỉ để nạp (ingest) dữ liệu trước khi có thể chạy một câu truy vấn duy nhất.

Phân tích nguyên nhân gốc rễ: Chiếm dụng bộ nhớ và hạn chế của cơ chế Row-Oriented

Tại sao các công cụ xử lý dữ liệu tiêu chuẩn lại chật vật khi tập dữ liệu vượt qua ngưỡng vài gigabyte? Vấn đề nằm ở cách cấu trúc dữ liệu của Python và các engine hướng dòng (row-oriented) vận hành bên dưới:

  • Chi phí quá tải từ Python Object: Pandas tiêu chuẩn lưu trữ chuỗi dưới dạng các con trỏ Python object thông thường trừ khi bạn chủ động cấu hình PyArrow engine. Mỗi cell dữ liệu sẽ kéo theo khoảng 40 đến 60 byte metadata phụ trợ. Một file CSV 1.2 GB có thể dễ dàng phình to thành 5.8 GB RAM vật lý.
  • Cơ chế sao chép phòng thủ (Defensive Copying): Các biểu thức lọc như df[df['status'] == 'completed'] mặc định sẽ cấp phát hoàn toàn các DataFrame mới. Việc nối liên tiếp ba thao tác lọc sẽ khiến lượng RAM tiêu thụ tăng gấp ba lần trừ khi bạn tự quản lý view và garbage collection thủ công.
  • Xử lý dạng dòng (Row-Based) so với dạng cột (Columnar): SQLite lưu trữ các hàng liên tục nhau. Nếu bạn tính AVG(latency) trên 40 triệu bản ghi, SQLite sẽ đọc toàn bộ timestamp, IP address và payload từ ổ đĩa chỉ để lấy ra duy nhất một giá trị số nguyên. Việc đó gây lãng phí disk I/O và làm tràn bộ nhớ đệm CPU với các thuộc tính không cần thiết.
  • Nút thắt xử lý đơn luồng: Cả SQLite lẫn lõi Pandas đều thực thi các thao tác phân tích trên một core duy nhất. Trong khi một thread chạm đỉnh 100%, 7 hoặc 15 core còn lại trên CPU của bạn vẫn hoàn toàn rảnh rỗi.

So sánh các giải pháp: Pandas, SQLite, Spark và DuckDB

Khi bạn cần phân tích từ 10 triệu đến 100 triệu dòng trên máy cá nhân của lập trình viên, có bốn lựa chọn phổ biến:

1. Pandas / Xử lý theo Chunk

Pandas hoạt động rất tốt khi dữ liệu nằm gọn trong mức 20% dung lượng RAM hệ thống. Vượt quá giới hạn đó, việc thiếu execution planner sẽ bộc lộ điểm yếu. Logic chia chunk rất dễ phát sinh lỗi và biến các phép tổng hợp đơn giản thành cơn ác mộng bảo trì.

2. SQLite

SQLite cực kỳ ổn định và đáng tin cậy cho việc lưu trữ ứng dụng mang tính giao dịch (transactional). Tuy nhiên, việc chạy tính tổng theo nhóm trên 50 triệu bản ghi sẽ mất hàng phút thay vì vài giây vì nó thiếu khả năng quét song song và lưu trữ dạng cột.

3. Apache Spark cục bộ / PySpark

Spark xử lý khối lượng công việc khổng lồ mà không hề nao núng. Nhưng việc chạy nó cục bộ trên một máy đơn lẻ lại rất cồng kềnh. Bạn phải gánh thêm chi phí vận hành của JVM, mất 15 giây chỉ để khởi động executor, kèm theo hàng loạt tham số tinh chỉnh bộ nhớ phức tạp chỉ cho một script phân tích thăm dò.

4. DuckDB

DuckDB lấp đầy khoảng trống này. Hãy hình dung nó giống như một SQLite được xây dựng chuyên biệt cho mục đích phân tích dữ liệu. Nó chạy in-process dưới dạng engine C++ nhúng mà không cần bất kỳ background service nào. Bên dưới hệ thống, DuckDB sử dụng cơ chế thực thi dạng cột theo vector (vectorized columnar execution), tận dụng tối đa tất cả các CPU core và có thể stream các tập dữ liệu lớn hơn RAM trực tiếp từ ổ cứng.

Cách tiếp cận tối ưu: Triển khai thực tế với DuckDB và Python

Mô hình gọn gàng nhất là kết hợp trực tiếp SQL engine hướng vector của DuckDB với Python. Bạn quét các file trực tiếp trên ổ cứng tại chỗ, xử lý dữ liệu bằng C++ và chỉ hiện thực hóa (materialize) kết quả tổng hợp cuối cùng thành một DataFrame của Pandas hoặc Polars.

1. Cài đặt

DuckDB cung cấp các bản pre-compiled wheel hoàn chỉnh mà không cần dependency ngoài. Hãy cài đặt bộ công cụ thông qua pip:

pip install duckdb pyarrow pandas

2. Truy vấn trực tiếp các file thô mà không cần nạp vào Database

DuckDB tự động kiểm tra header và truy vấn file trực tiếp. Bạn có thể bỏ qua hoàn toàn các bước khai báo bảng và nạp dữ liệu thủ công:

import duckdb

# Truy vấn 25 triệu bản ghi trên các file Parquet đã phân vùng
query = """
    SELECT 
        merchant_id,
        COUNT(*) AS total_transactions,
        ROUND(AVG(amount), 2) AS avg_amount,
        ROUND(SUM(amount), 2) AS total_revenue
    FROM 'data/transactions_*.parquet'
    WHERE status = 'SUCCESS'
    GROUP BY merchant_id
    HAVING COUNT(*) > 1000
    ORDER BY total_revenue DESC
    LIMIT 10;
"""

# DuckDB đọc stream các footer của parquet và trả về kết quả trong ~0.8s
results = duckdb.sql(query)
results.show()

3. Tích hợp Zero-Copy với Python DataFrame

DuckDB chia sẻ bộ đệm bộ nhớ trực tiếp với Apache Arrow và Pandas. Nó đẩy các tác vụ tính toán nặng xuống engine bên dưới và chỉ trả về các dòng kết quả tóm tắt gọn nhẹ mà không phát sinh chi phí serialization:

import duckdb

con = duckdb.connect(database=":memory:")

# Xử lý cô đọng 40 triệu dòng xuống còn 500 dòng, sau đó xuất sang Pandas
summary_df = con.execute("""
    SELECT 
        DATE_TRUNC('day', timestamp) AS tx_date,
        country_code,
        COUNT(DISTINCT user_id) AS active_users,
        SUM(amount) AS gross_volume
    FROM 'data/raw_access_logs.parquet'
    GROUP BY 1, 2
""").df()

print(f"Kích thước dữ liệu tóm tắt sau xử lý: {summary_df.shape}")
print(summary_df.head())

4. Xử lý tập dữ liệu lớn hơn RAM (Out-of-Core Processing)

Điều gì xảy ra khi bạn cần truy vấn một file clickstream dung lượng 80 GB trên một chiếc laptop chỉ có 16 GB RAM? DuckDB tự động điều phối cơ chế tràn bộ nhớ (memory-spilling). Bạn có thể chủ động giới hạn mức RAM tiêu thụ và chuyển hướng các file tạm sang ổ lưu trữ NVMe tốc độ cao:

import duckdb

# Giới hạn mức sử dụng RAM và cấu hình ghi tràn dữ liệu ra đĩa
con = duckdb.connect(database="analytics_cache.duckdb")

con.execute("SET max_memory = '10GB';")
con.execute("SET temp_directory = '/tmp/duckdb_spill';")
con.execute("SET threads = 8;")

# Xử lý out-of-core và lưu file Parquet đã nén ngược lại ổ đĩa
con.execute("""
    COPY (
        SELECT 
            user_id,
            COUNT(*) AS event_count,
            APPROX_COUNT_DISTINCT(session_id) AS approx_sessions
        FROM 'data/huge_clickstream_dump.csv'
        GROUP BY user_id
    ) TO 'output/user_behavior_metrics.parquet' (FORMAT PARQUET, COMPRESSION ZSTD);
""")

con.close()

Kinh nghiệm thực chiến từ môi trường Production

Dưới đây là ba bài học thực tế đắt giá rút ra từ việc ứng dụng DuckDB trong các ETL script hàng ngày và các scheduled job:

  • Nên ưu tiên Parquet hơn CSV: DuckDB phân tích cú pháp CSV rất nhanh nhờ cơ chế parallel sniffer. Tuy nhiên, định dạng Parquet có sẵn metadata thống kê cột và byte offset. Điều này cho phép DuckDB bỏ qua toàn bộ các block dữ liệu không liên quan khi tính toán bộ lọc WHERE, giúp giảm thời gian truy vấn trên tập dữ liệu 20 GB từ 42 giây xuống chỉ còn 3.5 giây.
  • Truy vấn trực tiếp các Pandas DataFrame trong bộ nhớ: DuckDB xem bất kỳ DataFrame nào đang hoạt động trong scope của Python như một bảng SQL in-memory. Bạn không cần phải chuyển đổi thủ công: câu lệnh duckdb.sql("SELECT * FROM my_df WHERE score > 90") có thể hoạt động được ngay.
  • Lưu ý hiện tượng tranh chấp File Lock: Một file DuckDB hỗ trợ nhiều tiến trình đọc đồng thời, nhưng chỉ cho phép duy nhất một tiến trình ghi tại một thời điểm. Nếu pipeline của bạn chạy nhiều worker process song song, hãy để mỗi worker ghi ra một file tạm độc lập, sau đó tổng hợp chúng lại thông qua một unified view.

Quy trình làm việc này mang lại hiệu quả vượt trội. Với pipeline 35 GB của chúng tôi, việc thay thế các script chia chunk thủ công và các container database dùng tạm thời bằng DuckDB đã rút ngắn thời gian chạy pipeline từ 18 phút xuống còn 46 giây. Mã nguồn thu được ngắn hơn, hoạt động tin cậy hơn và hoàn toàn không tốn công bảo trì hạ tầng.

Share:
Tags: