DuckDB Parquet 性能完全指南：比 CSV 快 10 倍的秘密

import duckdb

# 连接内存数据库
con = duckdb.connect(":memory:")

# 从 CSV 读取数据（自动推断类型）
con.execute("""
    CREATE TABLE orders AS
    SELECT * FROM read_csv_auto('/data/sales/orders.csv')
""")

# 快速导出为 Parquet（一步完成）
con.execute("""
    COPY orders TO '/data/sales/orders.parquet'
    (FORMAT PARQUET, COMPRESSION ZSTD)
""")

print("✅ 导出完成")

# 只读取需要的列，只读取满足条件的数据
result = con.execute("""
    SELECT customer_id, SUM(amount) AS total
    FROM read_parquet('/data/sales/*.parquet')
    WHERE order_date >= '2026-07-01'
      AND region = '华东'
    GROUP BY customer_id
    ORDER BY total DESC
    LIMIT 100
""").fetchdf()

print(result)

import time

# 测试 1：读取 1000 万行数据
print("=== 读取性能对比 ===")

# CSV 方式
start = time.time()
con.execute("SELECT * FROM read_csv_auto('/data/orders.csv')")
csv_time = time.time() - start
print(f"CSV 读取: {csv_time:.2f} 秒")

# Parquet 方式
start = time.time()
con.execute("SELECT * FROM read_parquet('/data/orders.parquet')")
parquet_time = time.time() - start
print(f"Parquet 读取: {parquet_time:.2f} 秒")
print(f"加速比: {csv_time/parquet_time:.1f}x")

# 测试 2：只读特定列
print("\n=== 列选择性能对比 ===")

# CSV 方式（必须读完整行）
start = time.time()
con.execute("SELECT customer_id, amount FROM read_csv_auto('/data/orders.csv')")
csv_col_time = time.time() - start
print(f"CSV 选列: {csv_col_time:.2f} 秒")

# Parquet 方式（只读需要的列）
start = time.time()
con.execute("SELECT customer_id, amount FROM read_parquet('/data/orders.parquet')")
parquet_col_time = time.time() - start
print(f"Parquet 选列: {parquet_col_time:.2f} 秒")
print(f"加速比: {csv_col_time/parquet_col_time:.1f}x")

-- 按年份和月份分区写入
COPY orders TO '/data/sales/parquet_partitioned/'
(
    FORMAT PARQUET,
    PARTITION_BY (order_date YEAR, order_date MONTH)
);

-- 只扫描 2026 年 7 月的数据，其他分区自动跳过
SELECT *
FROM read_parquet('/data/sales/parquet_partitioned/')
WHERE order_date >= '2026-07-01'
  AND order_date < '2026-08-01';

import duckdb
import pandas as pd
from pathlib import Path

class DataPipeline:
    """数据管道：CSV 输入 → Parquet 处理 → 分析输出"""
    
    def __init__(self, input_dir, output_dir):
        self.con = duckdb.connect(":memory:")
        self.input_dir = Path(input_dir)
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(parents=True, exist_ok=True)
    
    def ingest_csv(self, filename):
        """导入 CSV 并转为 Parquet"""
        csv_path = self.input_dir / filename
        table_name = Path(filename).stem
        
        # 读取 CSV
        self.con.execute(f"""
            CREATE TABLE {table_name} AS
            SELECT * FROM read_csv_auto('{csv_path}')
        """)
        
        # 导出 Parquet
        parquet_path = self.output_dir / f"{table_name}.parquet"
        self.con.execute(f"""
            COPY {table_name} TO '{parquet_path}'
            (FORMAT PARQUET, COMPRESSION ZSTD)
        """)
        
        print(f"✅ {filename} → {parquet_path}")
        return parquet_path
    
    def query_parquet(self, sql):
        """查询 Parquet 文件"""
        return self.con.execute(sql).fetchdf()
    
    def export_results(self, df, filename):
        """导出结果"""
        output_path = self.output_dir / filename
        df.to_csv(output_path, index=False)
        print(f"✅ 结果已保存到 {output_path}")
        return output_path

# 使用示例
pipeline = DataPipeline('/data/input', '/data/output')

# 导入数据
pipeline.ingest_csv('orders.csv')
pipeline.ingest_csv('customers.csv')

# 分析查询
result = pipeline.query_parquet("""
    SELECT 
        c.customer_name,
        COUNT(o.order_id) AS order_count,
        ROUND(SUM(o.amount), 2) AS total_amount
    FROM read_parquet('/data/output/customers.parquet') c
    JOIN read_parquet('/data/output/orders.parquet') o
        ON c.customer_id = o.customer_id
    GROUP BY c.customer_name
    ORDER BY total_amount DESC
    LIMIT 10
""")

# 导出结果
pipeline.export_results(result, 'top_customers.csv')