DuckDB + Python 数据科学工作流

import duckdb# 1. 连接（默认是内存数据库）con = duckdb.connect()             # 内存数据库con = duckdb.connect('my.duckdb')  # 持久化数据库# 2. 执行查询（两种写法）result = duckdb.sql("SELECT 42")     # 模块级（用全局连接）result = con.sql("SELECT 42")        # 连接级# 3. 获取结果（多种格式）result.show()          # 直接打印表格result.fetchall()      # 返回 Python listresult.df()            # 返回 Pandas DataFrameresult.pl()            # 返回 Polars DataFrameresult.arrow()         # 返回 Arrow Tableresult.fetchnumpy()    # 返回 NumPy 字典# 4. 直接读文件duckdb.read_csv('data.csv')duckdb.read_parquet('data.parquet')duckdb.read_json('data.json')

┌───────────────────────────────────────────────────┐│                 你的 Python 脚本                    ││                                                    ││  ┌──────────┐   ┌──────────┐   ┌──────────┐       ││  │  Pandas   │   │  Polars  │   │  Arrow   │       ││  │ DataFrame │   │DataFrame │   │  Table   │       ││  └─────┬────┘   └────┬─────┘   └────┬─────┘       ││        │             │              │              ││        └──────┬──────┘──────────────┘              ││               │ (零拷贝 / 自动检测)                   ││               ▼                                    ││      ┌────────────────┐                            ││      │    DuckDB SQL    │ ←── SQL 查询              ││      │   Engine         │                          ││      └───────┬────────┘                            ││              │                                     ││     ┌────────┼─────────┐                           ││     ▼        ▼         ▼                           ││  .csv     .parquet   .json    ← 直接查文件          │└───────────────────────────────────────────────────┘

import duckdbimport pandas as pd# 创建两个 DataFrameorders = pd.DataFrame({    'order_id': [1, 2, 3, 4, 5],    'customer_id': [101, 102, 101, 103, 102],    'amount': [250, 130, 80, 420, 310],    'status': ['shipped', 'pending', 'shipped', 'shipped', 'cancelled']})customers = pd.DataFrame({    'id': [101, 102, 103],    'name': ['Alice', 'Bob', 'Charlie'],    'city': ['Beijing', 'Shanghai', 'Shenzhen']})# 直接用 SQL 查询 DataFrame！result = duckdb.sql("""    SELECT c.name, c.city,           COUNT(*) AS order_count,           SUM(o.amount) AS total_spent    FROM orders o    JOIN customers c ON o.customer_id = c.id    WHERE o.status = 'shipped'    GROUP BY c.name, c.city    ORDER BY total_spent DESC""").df()print(result)

      name      city  order_count  total_spent0  Charlie  Shenzhen            1          4201    Alice   Beijing            2          330

import duckdbimport time# ---- DuckDB 方式 ----start = time.time()result = duckdb.sql("""    SELECT         DATE_TRUNC('month', event_date::DATE) AS month,        event_type,        COUNT(*) AS event_count,        COUNT(DISTINCT user_id) AS unique_users    FROM 'large_events.csv'    GROUP BY ALL    ORDER BY month, event_count DESC""").df()print(f"DuckDB: {time.time() - start:.2f}s, 结果 {len(result)} 行")# ---- Pandas 方式（对比） ----import pandas as pdstart = time.time()df = pd.read_csv('large_events.csv', parse_dates=['event_date'])result2 = (df.groupby([df.event_date.dt.to_period('M'), 'event_type'])             .agg(event_count=('event_type', 'count'),                  unique_users=('user_id', 'nunique'))             .sort_values(['event_date', 'event_count'], ascending=[True, False])             .reset_index())print(f"Pandas: {time.time() - start:.2f}s, 结果 {len(result2)} 行")

import duckdbresult = duckdb.sql("""    SELECT         user_id,        order_date,        amount,        -- 每个用户内部按金额排名        RANK() OVER (PARTITION BY user_id ORDER BY amount DESC) AS amount_rank,        -- 与该用户上一笔订单的天数间隔        DATEDIFF('day',                 LAG(order_date) OVER (PARTITION BY user_id ORDER BY order_date),                 order_date        ) AS days_since_last_order,        -- 该用户的累计消费        SUM(amount) OVER (PARTITION BY user_id ORDER BY order_date) AS cumulative_spent    FROM 'orders.csv'    ORDER BY user_id, order_date""").df()print(result.head(10))

# Cell 1: 初始化import duckdbcon = duckdb.connect()# Cell 2: 查看数据概览con.sql("""    SELECT * FROM 'sales_data.parquet'     USING SAMPLE 5 ROWS""")# Jupyter 会自动把 Relation 渲染成漂亮的表格# Cell 3: 分析monthly = con.sql("""    SELECT DATE_TRUNC('month', sale_date) AS month,           SUM(revenue) AS total_revenue,           COUNT(DISTINCT customer_id) AS unique_customers    FROM 'sales_data.parquet'    GROUP BY 1    ORDER BY 1""").df()# Cell 4: 可视化（用回 Pandas/Matplotlib）import matplotlib.pyplot as pltmonthly.plot(x='month', y='total_revenue', kind='bar', figsize=(12, 5))plt.title('Monthly Revenue')plt.tight_layout()plt.show()

import duckdbimport pandas as pdimport numpy as npimport timedef benchmark(n_rows):    """生成 n_rows 行数据，对比 DuckDB vs Pandas 的聚合性能"""    # 生成测试数据    np.random.seed(42)    df = pd.DataFrame({        'category': np.random.choice(['A','B','C','D','E'], n_rows),        'region': np.random.choice(['East','West','North','South'], n_rows),        'value': np.random.randn(n_rows) * 100,        'count': np.random.randint(1, 100, n_rows)    })    # DuckDB    start = time.time()    duckdb.sql("""        SELECT category, region,                SUM(value), AVG(value), COUNT(*), MAX(count)        FROM df GROUP BY category, region    """).df()    t_duck = time.time() - start    # Pandas    start = time.time()    df.groupby(['category', 'region']).agg({        'value': ['sum', 'mean', 'count'],        'count': 'max'    }).reset_index()    t_pandas = time.time() - start    return t_duck, t_pandas# 运行测试for n in [100_000, 1_000_000, 10_000_000]:    t_duck, t_pandas = benchmark(n)    speedup = t_pandas / t_duck    print(f"{n:>12,} 行 | DuckDB: {t_duck:.3f}s | Pandas: {t_pandas:.3f}s | 加速比: {speedup:.1f}x")