Khoa học Dữ liệu Tốc độ Cao: DuckDB và Polars cho Phân tích Hiệu suất Cao

Dataset on Disk (Parquet/CSV): 2.0 GB
Pandas in RAM:                12.0 GB - 18.0 GB (Peak during merge/groupby)
Polars in RAM:                 2.3 GB (Zero-copy Arrow + streaming)
DuckDB in RAM:                 0.8 GB (Chunked vectorized execution engine)

import polars as pl

# Construct lazy query plan — zero disk I/O occurs here
lazy_query = (
    pl.scan_parquet("s3://analytics-bucket/events/*.parquet")
    .filter(pl.col("timestamp") >= pl.date(2026, 1, 1))
    .filter(pl.col("event_type").is_in(["purchase", "subscription"]))
    .with_columns([
        (pl.col("amount_cents") / 100.0).alias("amount_usd"),
        pl.col("user_id").n_unique().over("country").alias("unique_users_per_country")
    ])
    .group_by(["country", "event_type"])
    .agg([
        pl.col("amount_usd").sum().alias("total_revenue"),
        pl.col("amount_usd").mean().alias("avg_order_value"),
        pl.len().alias("transaction_count")
    ])
    .sort("total_revenue", descending=True)
)

# Inspect the optimized query plan
print(lazy_query.explain())

# Execute optimized plan in parallel across all CPU cores
result_df = lazy_query.collect(streaming=True)
print(result_df)

import duckdb

# Connect to in-process DuckDB instance (or persist to 'analytics.duckdb')
con = duckdb.connect()

# Enable S3 / HTTP filesystem extension
con.execute("INSTALL httpfs; LOAD httpfs;")
con.execute("""
    SET s3_region='us-east-1';
    SET s3_access_key_id='YOUR_KEY';
    SET s3_secret_access_key='YOUR_SECRET';
""")

# Query 100GB of remote Parquet files using Vectorized SQL
query = """
    SELECT 
        country,
        event_type,
        COUNT(DISTINCT user_id) AS unique_users,
        ROUND(SUM(amount_cents) / 100.0, 2) AS total_revenue_usd,
        ROUND(AVG(amount_cents) / 100.0, 2) AS aov_usd
    FROM read_parquet('s3://analytics-bucket/events/year=2026/*/*.parquet')
    WHERE event_type IN ('purchase', 'subscription')
    GROUP BY country, event_type
    HAVING total_revenue_usd > 10000
    ORDER BY total_revenue_usd DESC
    LIMIT 20;
"""

# Execute and fetch directly to Arrow, Polars, or Python dictionaries
results = con.execute(query).pl()  # Returns native Polars DataFrame
print(results)

import polars as pl
import duckdb

# 1. Load and clean data with Polars
df = pl.DataFrame({
    "user_id": [101, 102, 103, 104],
    "scores": [88.5, 92.0, 79.5, 95.0],
    "tier": ["gold", "platinum", "gold", "platinum"]
})

# 2. Run complex analytical window SQL in DuckDB directly on the Polars DataFrame
con = duckdb.connect()

# DuckDB can reference the 'df' Python variable directly in the FROM clause!
sql_result = con.execute("""
    SELECT 
        user_id,
        scores,
        tier,
        RANK() OVER (PARTITION BY tier ORDER BY scores DESC) as rank_in_tier,
        AVG(scores) OVER (PARTITION BY tier) as tier_avg
    FROM df
""").arrow()  # Zero-copy Arrow Table

# 3. Convert back to Polars instantaneously
final_df = pl.from_arrow(sql_result)
print(final_df)

                                Dataset Scale & Use Case
                                           │
         ┌─────────────────────────────────┴─────────────────────────────────┐
         ▼                                                                   ▼
  Single Machine (< 500GB)                                          Distributed Cluster (> 1TB)
         │                                                                   │
    ┌────┴──────────────────────────┐                              ┌─────────┴─────────┐
    ▼                               ▼                              ▼                   ▼
SQL-Heavy / S3 Parquet       DataFrame Transformations      Batch Pipeline       Real-time OLAP
    ▼                               ▼                              ▼                   ▼
 DuckDB                          Polars                       Apache Spark         ClickHouse
 (Embedded SQL)              (Rust Multi-thread)               / Ray               / StarRocks