DuckDB 生产环境性能优化：从慢查询到秒级响应的完整链路

WITH cleaned AS (
    SELECT * FROM read_csv_auto('orders.csv')
    WHERE total_amount > 0
),
filtered AS (
    SELECT * FROM cleaned
    WHERE order_time >= '2024-01-01'
),
enriched AS (
    SELECT f.*, u.region
    FROM filtered f
    LEFT JOIN users u ON f.user_id = u.id
)
SELECT category, COUNT(*) as cnt, SUM(total_amount) as gmv
FROM enriched
GROUP BY category;

import duckdb

conn = duckdb.connect('ecommerce.db')

# 第一步：先把清洗后的宽表物化为 Parquet（一次写入，多次读取）
conn.execute("""
    CREATE TABLE clean_orders AS
    SELECT 
        order_id,
        user_id,
        order_time,
        total_amount,
        category,
        region
    FROM read_csv_auto('orders.csv')
    WHERE total_amount > 0
      AND order_time >= '2024-01-01'
""")

# 第二步：导出为 Parquet，后续查询直接读文件
conn.execute("COPY clean_orders TO 'data/clean_orders.parquet' (FORMAT PARQUET)")

# 第三步：后续分析直接读 Parquet
result = conn.execute("""
    SELECT category, 
           COUNT(*) as order_cnt,
           SUM(total_amount) as gmv,
           AVG(total_amount) as avg_order
    FROM read_parquet('data/clean_orders.parquet')
    GROUP BY category
    ORDER BY gmv DESC
""").fetchdf()

import duckdb

conn = duckdb.connect('analysis.db')

# 场景 A：分析机只跑 DuckDB，独占 CPU
conn.execute("SET threads TO 8")  # 明确指定，避免动态切换开销

# 场景 B：服务器上有其他进程，限制 DuckDB 用量
conn.execute("SET threads TO 4")
conn.execute("SET memory_limit TO '8GB'")  # 防止 OOM

# 场景 C：单用户笔记本，让 DuckDB 用满核心
conn.execute("SET threads TO 0")  # 0 表示使用所有可用核心

# 关键：并行度越高，内存占用越大
conn.execute("SET max_threads TO 4")
conn.execute("SET memory_limit TO '16GB'")
conn.execute("SET temp_directory TO '/tmp/duckdb_temp'")  # 溢出到磁盘，避免内存爆炸

SET enable_projection_pushdown TO true;
SET enable_multiphase_agg TO true;

# ❌ 错误示范：循环写入，每次都是独立事务
for chunk in pd.read_csv('huge_data.csv', chunksize=10000):
    conn.execute("INSERT INTO target_table VALUES ?", [tuple(row) for row in chunk.itertuples(index=False)])

# ✅ 方案1：用 COPY FROM（最快，直接走列存路径）
conn.execute("COPY target_table FROM 'huge_data.csv' (FORMAT CSV, HEADER)")

# ✅ 方案2：Pandas 批量插入（适合需要预处理的数据）
import pandas as pd
df = pd.read_csv('huge_data.csv')
with conn.transaction():
    for batch in pd.concat([df[i:i+50000] for i in range(0, len(df), 50000)]):
        batch.to_sql('target_table', conn, if_exists='append', index=False, method='multi')

# ✅ 方案3：直接用 DuckDB 的 read_csv + CTAS（最适合纯分析场景）
conn.execute("""
    CREATE TABLE target_table AS
    SELECT * FROM read_csv_auto('huge_data.csv')
""")

import duckdb

# 正确的目录结构（DuckDB 自动识别分区）
# data/sales/year=2024/month=01/part-000.parquet
# data/sales/year=2024/month=02/part-000.parquet
# data/sales/year=2025/month=01/part-000.parquet

conn = duckdb.connect('sales.db')

# DuckDB 会自动识别分区路径，只读取 2025 年的数据
result = conn.execute("""
    SELECT month, SUM(revenue) as total
    FROM read_parquet('data/sales/**/*.parquet')
    WHERE year = 2025
    GROUP BY month
    ORDER BY total DESC
""").fetchdf()

-- 强制开启分区裁剪（某些情况下优化器会失效）
SET allow_partitioned_scan_pruning TO true;

conn = duckdb.connect('warehouse.db')

# 创建逻辑视图（不存储数据，只存定义）
conn.execute("""
    CREATE VIEW v_user_30d AS
    SELECT 
        u.user_id,
        u.registration_date,
        COUNT(o.order_id) as order_cnt_30d,
        SUM(o.total_amount) as gmv_30d
    FROM users u
    LEFT JOIN orders o ON u.user_id = o.user_id
        AND o.order_time >= CURRENT_DATE - INTERVAL 30 DAY
    GROUP BY u.user_id, u.registration_date
""")

# 创建物化视图（存储结果）
conn.execute("""
    CREATE MATERIALIZED VIEW mv_user_30d AS
    SELECT * FROM v_user_30d
""")

# 定期刷新物化视图（比重新计算快得多）
conn.execute("REFRESH MATERIALIZED VIEW mv_user_30d")

# 后续所有查询都读物化视图，而不是重新计算
result = conn.execute("""
    SELECT * FROM mv_user_30d 
    WHERE gmv_30d > 1000 
    ORDER BY gmv_30d DESC 
    LIMIT 100
""").fetchdf()