Python 深度集成

5.1 连接方式：内存 vs 持久化

PYTHONimport duckdb

# 方式1：内存数据库（进程退出后数据消失）
con = duckdb.connect()                    # 内存
# 或
con = duckdb.connect(':memory:')          # 等价写法

# 方式2：持久化到文件
con = duckdb.connect('analytics.duckdb') # 自动创建文件

# 使用 with 语句自动关闭
with duckdb.connect('analytics.duckdb') as con:
    result = con.execute("SELECT 42").fetchone()

# 模块级默认连接（最简单，适合脚本）
duckdb.sql("SELECT 'hello'").show()

5.2 执行查询与获取结果

PYTHONcon = duckdb.connect()

# fetchall() → list of tuples
rows = con.execute("SELECT 1, 'a'").fetchall()

# fetchone() → 单行 tuple
row = con.execute("SELECT COUNT(*) FROM 'data.csv'").fetchone()
count = row[0]

# df() → pandas DataFrame（最常用！）
df = con.execute("""
    SELECT region, SUM(amount) as total
    FROM 'sales.csv'
    GROUP BY region
""").df()

# fetchdf() 等价于 df()
df = con.execute("SELECT * FROM 'data.parquet'").fetchdf()

# pl() → polars DataFrame（需要安装 polars）
import polars as pl
df_pl = con.execute("SELECT * FROM 'data.parquet'").pl()

# arrow() → Apache Arrow Table（零拷贝转换）
arrow_table = con.execute("SELECT * FROM 'data.parquet'").arrow()

# .show() → 终端打印（CLI 风格）
con.execute("SELECT * FROM 'data.csv' LIMIT 5").show()

5.3 零拷贝操作 pandas DataFrame

DuckDB 最强大的特性之一：直接在 SQL 中引用 Python 变量名作为表名，无需显式注册，数据不复制（Zero-Copy）。

PYTHONimport pandas as pd
import duckdb

# 创建 pandas DataFrame
df = pd.DataFrame({
    'name': ['Alice', 'Bob', 'Carol'],
    'age': [25, 30, 28],
    'score': [88.5, 92.0, 76.3]
})

# 直接在 SQL 中用变量名 "df" 引用！
result = duckdb.sql("SELECT * FROM df WHERE age > 26 ORDER BY score DESC").df()

# 合并多个 DataFrame
df2 = pd.read_csv('extra_data.csv')
combined = duckdb.sql("""
    SELECT * FROM df
    UNION ALL
    SELECT * FROM df2
""").df()

# 用 SQL 替代复杂的 pandas 操作
pivot_result = duckdb.sql("""
    SELECT name, SUM(score) as total_score
    FROM df
    GROUP BY name
    HAVING total_score > 80
    ORDER BY total_score DESC
""").df()

5.4 操作 polars DataFrame

PYTHONimport polars as pl
import duckdb

lazy_df = pl.scan_parquet('large_data.parquet')  # Polars 惰性求值

# DuckDB 可以直接查询 Polars LazyFrame
result = duckdb.sql("""
    SELECT category, COUNT(*) as cnt, AVG(price) as avg_price
    FROM lazy_df
    GROUP BY category
    ORDER BY cnt DESC
""").pl()  # 返回 Polars DataFrame

print(result)

5.5 注册 Python 对象为视图

PYTHONcon = duckdb.connect()

# 注册 DataFrame 为永久视图名
con.register('sales_view', sales_df)
con.execute("SELECT * FROM sales_view LIMIT 5")

# 注册后可以在多个查询中使用
q1 = con.execute("SELECT COUNT(*) FROM sales_view").fetchone()[0]
q2 = con.execute("SELECT SUM(amount) FROM sales_view").fetchone()[0]

# 参数化查询（防 SQL 注入）
product = '手机'
result = con.execute(
    "SELECT * FROM sales_view WHERE product = ?",
    [product]
).df()

5.6 Relation API 链式查询

Relation API 是 DuckDB 的 Python 原生 API，类似 pandas 的链式操作，构建查询而不直接执行。

PYTHONcon = duckdb.connect()

# 从 CSV 创建 Relation
rel = con.read_csv('sales.csv')

# 链式操作
result = (
    con.read_csv('sales.csv')
      .filter("amount > 1000")
      .aggregate("region, SUM(amount) as total", "region")
      .order("total DESC")
      .limit(10)
      .df()
)

# 查看生成的 SQL
rel.query('rel', "SELECT * FROM rel LIMIT 5").show()

# Relation 支持 JOIN
orders_rel = con.read_csv('orders.csv')
users_rel  = con.read_csv('users.csv')
joined = orders_rel.join(users_rel, "user_id")  # 自然 JOIN

5.7 Jupyter Notebook 中使用

PYTHON# 在 Jupyter 单元格中的推荐用法
import duckdb
import pandas as pd

# 推荐：用 duckdb.sql() 直接查询，结果用 .df() 转 pandas
df = duckdb.sql("""
    SELECT
        date_trunc('month', order_date) as month,
        product_category,
        SUM(amount) as revenue,
        COUNT(*) as orders
    FROM 'orders.parquet'
    WHERE order_date >= '2024-01-01'
    GROUP BY ALL   -- DuckDB 特性：自动按非聚合列 GROUP BY
    ORDER BY month, revenue DESC
""").df()

# 直接展示（Rich 格式输出，带列名和类型）
duckdb.sql("SELECT * FROM df LIMIT 5")  # Jupyter 中自动渲染

# 安装 jupysql 扩展（支持 %%sql 魔法命令）
# pip install jupysql
# %load_ext sql
# %sql duckdb:///:memory:
# %%sql
# SELECT * FROM 'data.csv' LIMIT 5

5.8 实战：用 DuckDB 加速 pandas 慢查询

pandas 的某些操作在大数据量下非常慢，DuckDB 可以提速 10-100 倍：

PYTHONimport pandas as pd
import duckdb
import time

# 假设 df 有 500 万行
df = pd.read_parquet('large_events.parquet')

# ❌ pandas 做法（慢）
start = time.time()
result_pandas = (
    df[df['event_type'] == 'purchase']
    .groupby(['user_id', pd.Grouper(key='event_time', freq='D')])
    ['amount'].sum()
    .reset_index()
)
print(f"pandas: {time.time()-start:.2f}s")  # 可能 30s+

# ✅ DuckDB 做法（快）
start = time.time()
result_duck = duckdb.sql("""
    SELECT
        user_id,
        date_trunc('day', event_time) AS event_date,
        SUM(amount) AS daily_spend
    FROM df
    WHERE event_type = 'purchase'
    GROUP BY user_id, event_date
    ORDER BY user_id, event_date
""").df()
print(f"DuckDB: {time.time()-start:.2f}s")  # 通常 1-3s