DuckDB 完全指南

官方文档：https://duckdb.org/docs/
适用版本：DuckDB 1.5.2（2026-05-07 核实）

概述

DuckDB 是一个嵌入式列式 OLAP 数据库，无需独立服务进程，像 SQLite 一样嵌入应用内部运行，但专门针对分析型查询（大量数据扫描、聚合、连接）进行了优化。

与主要替代方案的对比：

适合的场景：

• 本地数据分析、数据探索（替代 Pandas 处理大文件）
• ETL 管道中的中间层处理
• 读取并分析 Parquet / CSV / JSON 文件，无需数据库服务器
• 需要 SQL 但不想维护 ClickHouse 服务的场景
• Jupyter Notebook 数据分析工作流

不适合的场景：

• 高并发 OLTP 写入（DuckDB 针对分析读，不是高频写）
• 需要多进程并发写入同一数据库文件
• 需要实时流处理（考虑 Kafka + ClickHouse）

安装

# 基础安装
pip install duckdb

# 含 Pandas 集成
pip install duckdb pandas

# 含 Polars 集成
pip install duckdb polars

# 含 PyArrow 集成
pip install duckdb pyarrow

核心概念

连接模式

DuckDB 有三种连接模式：

模块级默认连接

duckdb.sql() 等模块级函数使用一个全局默认连接，等效于 duckdb.connect(":default:")。单文件脚本直接用模块级函数即可；需要隔离或多线程时才创建独立连接。

Relation（关系对象）

duckdb.sql() 等方法返回 DuckDBPyRelation，它是一个惰性求值的符号查询节点——不会立即执行查询，链式调用 .filter()、.limit() 等方法都只是修改查询计划。调用 .fetchall()、.to_df() 等输出方法时才真正执行。

API 详解

duckdb.connect()

创建数据库连接。

import duckdb

con = duckdb.connect(database=":memory:", read_only=False, config={})

config 常用选项：

# 生产环境：限制资源
con = duckdb.connect(
    database="analytics.db",
    config={
        "threads": 8,
        "memory_limit": "16GB",
        "temp_directory": "/tmp/duckdb",
    }
)

Connection — 查询执行方法

execute()

执行单条 SQL 语句，支持参数化查询。

con.execute(query, parameters=None)

返回值：DuckDBPyConnection（支持链式调用）

# 位置参数：使用 ? 占位符
con.execute(
    "INSERT INTO users (name, age) VALUES (?, ?)",
    ["Alice", 30]
)

# 命名参数：$name 格式
con.execute(
    "SELECT * FROM users WHERE name = $name",
    {"name": "Alice"}
)

# 链式：execute 后立即 fetchall
rows = con.execute("SELECT * FROM users LIMIT 10").fetchall()

executemany()

批量执行，对每组参数执行一次 SQL。

con.executemany(query, parameter_sequence)

users = [("Bob", 25), ("Carol", 35), ("Dave", 28)]
con.executemany("INSERT INTO users VALUES (?, ?)", users)

大批量插入应优先使用 INSERT INTO ... SELECT 或 COPY FROM，executemany 性能远低于批量 SQL。

Connection — 结果获取方法

fetchone()

返回下一行，无更多结果时返回 None。

row = con.execute("SELECT name FROM users").fetchone()
# row = ("Alice",) 或 None

fetchmany()

返回指定数量的行。

con.fetchmany(size=1)

fetchall()

返回所有剩余行，结果为 list[tuple]。

rows = con.execute("SELECT * FROM users").fetchall()
# [("Alice", 30), ("Bob", 25), ...]

fetchdf()

返回 pandas DataFrame。需要安装 pandas。

df = con.execute("SELECT * FROM users").fetchdf()
# 等价于 .to_df()

fetchnumpy()

返回 numpy 数组字典 dict[str, np.ndarray]，键为列名。

arrays = con.execute("SELECT age FROM users").fetchnumpy()
# {"age": array([30, 25, 35])}

pl()

返回 Polars DataFrame。需要安装 polars。

df = con.execute("SELECT * FROM users").pl()

arrow()

返回 PyArrow Table。需要安装 pyarrow。

tbl = con.execute("SELECT * FROM users").arrow()

Connection — 连接管理方法

cursor()

在同一底层连接上创建新的游标（Cursor），可用于并发查询。

cursor = con.cursor()
cursor.execute("SELECT ...")

close()

显式关闭连接，释放资源。通常通过上下文管理器自动关闭。

# 推荐：使用上下文管理器
with duckdb.connect("data.db") as con:
    result = con.execute("SELECT 42").fetchone()
# 超出 with 块后自动关闭

register() / unregister()

将 Python 对象（DataFrame、Arrow Table 等）注册为 DuckDB 虚拟视图。

con.register(view_name, python_object)
con.unregister(view_name)

import pandas as pd

df = pd.read_csv("data.csv")
con.register("df_view", df)
result = con.execute("SELECT count(*) FROM df_view").fetchone()
con.unregister("df_view")

Relation API

Relation API 通过链式调用构建惰性查询，适合复杂的数据处理管道。

创建 Relation

duckdb.sql() / con.sql()

执行 SELECT SQL 并返回 Relation。

duckdb.sql(query, alias='', params=None)

import duckdb

rel = duckdb.sql("SELECT * FROM 'data.parquet' WHERE year = 2024")
print(rel.columns)   # 列名列表
print(rel.dtypes)    # 列类型列表

duckdb.table() / con.table()

从已有表或视图创建 Relation。

duckdb.table(table_name)

duckdb.from_df() / con.from_df()

从 pandas DataFrame 创建 Relation（零拷贝）。

duckdb.from_df(df)

duckdb.from_arrow() / con.from_arrow()

从 PyArrow Table 或 RecordBatch 创建 Relation（零拷贝）。

duckdb.from_arrow(arrow_object)

Relation 转换方法（惰性求值）

filter() — WHERE 过滤

rel.filter(filter_expr)

rel = duckdb.sql("FROM sales").filter("amount > 1000 AND region = 'CN'")

project() / select() — SELECT 投影

rel.project(project_expr)

rel = duckdb.sql("FROM users").project("name, age, age * 2 AS double_age")

order() / sort() — ORDER BY

rel.order(order_expr)

rel = rel.order("amount DESC NULLS LAST, created_at ASC")

limit() — LIMIT / OFFSET

rel.limit(n, offset=0)

aggregate() — GROUP BY 聚合

rel.aggregate(aggr_expr, group_expr='')

rel = (
    duckdb.sql("FROM sales")
    .aggregate("region, sum(amount) AS total", "region")
    .order("total DESC")
)

join() — JOIN

rel.join(other_rel, condition, how='inner')

orders = duckdb.table("orders")
users = duckdb.table("users")
result = orders.join(users, "orders.user_id = users.id", how="left")

集合操作

rel.union(other_rel)      # UNION ALL
rel.intersect(other_rel)  # INTERSECT
rel.except_(other_rel)    # EXCEPT
rel.cross(other_rel)      # CROSS JOIN

Relation 输出方法（触发执行）

to_df() — 输出为 pandas DataFrame

df = rel.to_df()

to_arrow() — 输出为 PyArrow Table

tbl = rel.to_arrow()

to_csv() — 写出 CSV 文件

rel.to_csv(filename, **kwargs)

to_parquet() — 写出 Parquet 文件

rel.to_parquet(filename, **kwargs)

show() — 打印结果表格

rel.show(max_rows=None, max_width=None)

describe() — 描述性统计

rel.describe()  # 返回 count / mean / std / min / max / median

explain() — 查询计划

rel.explain(type='standard')

数据导入：read_csv()

直接读取 CSV 文件，可用于 SQL 中或作为 Relation 起点。

-- SQL 中直接读取
SELECT * FROM read_csv('data.csv', header=true, delim=',');

-- 读取目录下所有 CSV
SELECT * FROM read_csv('data/*.csv', union_by_name=true);

# Python Relation API
rel = duckdb.read_csv("data.csv", header=True)

完整参数表：

数据导入：read_parquet()

直接读取 Parquet 文件，支持 glob 通配符和 Hive 分区。

-- 读取单文件
SELECT * FROM read_parquet('data.parquet');

-- 读取目录下所有 Parquet（Hive 分区）
SELECT * FROM read_parquet('data/year=*/month=*/*.parquet', hive_partitioning=true);

rel = duckdb.read_parquet("data/year=2024/**/*.parquet", hive_partitioning=True)

数据导出：COPY TO

导出 CSV

COPY (SELECT * FROM sales WHERE year = 2024)
TO 'output.csv'
(FORMAT CSV, HEADER true, DELIMITER ',', COMPRESSION gzip);

导出 Parquet

COPY (SELECT * FROM large_table)
TO 'output.parquet'
(FORMAT PARQUET, COMPRESSION zstd, COMPRESSION_LEVEL 3, ROW_GROUP_SIZE 100000);

完整示例

场景一：本地 CSV 数据分析

import duckdb

# 读取并分析 CSV，无需加载到内存
result = duckdb.sql("""
    SELECT
        region,
        count(*) AS order_count,
        sum(amount) AS total_revenue,
        avg(amount) AS avg_order
    FROM read_csv('orders_2024.csv', header=true, auto_detect=true)
    WHERE status = 'completed'
    GROUP BY region
    ORDER BY total_revenue DESC
""").to_df()

print(result.head(10))

场景二：处理多个 Parquet 分区文件

import duckdb

con = duckdb.connect()

# 读取 Hive 分区 Parquet，合并 Schema 不同的文件
rel = con.read_parquet(
    "s3://bucket/data/year=*/month=*/*.parquet",
    hive_partitioning=True,
    union_by_name=True,
)

# 按分区聚合，过滤特定月份
monthly = (
    rel
    .filter("year = 2024")
    .aggregate("year, month, sum(revenue) AS total", "year, month")
    .order("year, month")
)

monthly.to_parquet("monthly_summary.parquet", compression="zstd")

场景三：与 Pandas 零拷贝协作

import duckdb
import pandas as pd

# 已有 Pandas DataFrame，直接交给 DuckDB 查询
df = pd.read_parquet("large_file.parquet")  # 假设 Pandas 能加载

# 零拷贝：DuckDB 直接引用 df 的内存
result = duckdb.sql("""
    SELECT
        category,
        percentile_cont(0.95) WITHIN GROUP (ORDER BY price) AS p95_price
    FROM df
    GROUP BY category
""").to_df()

场景四：ETL 管道：CSV → 清洗 → Parquet

import duckdb

with duckdb.connect("staging.db") as con:
    # 建表并导入
    con.execute("""
        CREATE OR REPLACE TABLE raw_events AS
        SELECT * FROM read_csv('events/*.csv',
            header=true,
            nullstr=['N/A', 'null', ''],
            normalize_names=true
        )
    """)

    # 清洗并导出
    con.execute("""
        COPY (
            SELECT
                event_id,
                user_id,
                strptime(event_time, '%Y-%m-%d %H:%M:%S') AS event_ts,
                lower(trim(event_type)) AS event_type,
                json_object('source', source, 'page', page) AS metadata
            FROM raw_events
            WHERE event_id IS NOT NULL
              AND user_id > 0
        )
        TO 'clean_events.parquet'
        (FORMAT PARQUET, COMPRESSION zstd, ROW_GROUP_SIZE 100000)
    """)
    print(f"Total rows: {con.execute('SELECT count(*) FROM raw_events').fetchone()[0]}")

场景五：持久化数据库的多步骤分析

import duckdb

with duckdb.connect("analytics.db") as con:
    # 创建或替换表
    con.execute("""
        CREATE OR REPLACE TABLE sales AS
        SELECT * FROM read_parquet('raw/*.parquet', hive_partitioning=true)
    """)

    # 创建视图
    con.execute("""
        CREATE OR REPLACE VIEW monthly_kpi AS
        SELECT
            date_trunc('month', sale_date) AS month,
            sum(revenue) AS revenue,
            count(DISTINCT customer_id) AS unique_customers,
            sum(revenue) / count(DISTINCT customer_id) AS arpu
        FROM sales
        GROUP BY 1
    """)

    # 查询视图
    df = con.sql("SELECT * FROM monthly_kpi ORDER BY month").to_df()

最佳实践

尽量直接查询文件而非先导入： DuckDB 能直接 SELECT Parquet/CSV 文件，对大文件只需要扫描不需要拷贝。当文件需要多次查询时再 CREATE TABLE AS SELECT 物化到数据库。

-- 对大型分析任务，先把需要的数据列和分区投影到 DuckDB 表
CREATE TABLE local_sales AS
SELECT region, product_id, amount, sale_date
FROM read_parquet('s3://bucket/sales/*.parquet', hive_partitioning=true)
WHERE sale_date >= '2024-01-01';
-- 之后的查询全部针对 local_sales，无需重复读 S3

使用上下文管理器管理连接： 用 with duckdb.connect(...) as con: 确保连接关闭，避免文件锁残留（持久化数据库文件只能有一个写入连接）。

# 正确
with duckdb.connect("data.db") as con:
    con.execute("INSERT INTO ...")

# 错误：忘记关闭会导致下次打开失败（文件被锁）
con = duckdb.connect("data.db")
con.execute("INSERT INTO ...")

并行读取多文件用 glob 而非循环： DuckDB 内置并行文件读取，一个 glob 表达式比 Python 循环拼接快数倍。

# 推荐：glob，DuckDB 并行读取所有文件
duckdb.sql("SELECT * FROM read_parquet('data/year=2024/*.parquet')")

# 不推荐：Python 循环，串行读取
import glob
for f in glob.glob("data/year=2024/*.parquet"):
    duckdb.sql(f"SELECT * FROM '{f}'")

参数化查询防止 SQL 注入： 动态构造查询时使用占位符，绝不拼接字符串。

user_id = request.args.get("user_id")

# 正确：参数化查询
result = con.execute(
    "SELECT * FROM users WHERE id = ?",
    [user_id]
).fetchall()

# 错误：字符串拼接，存在 SQL 注入风险
result = con.execute(f"SELECT * FROM users WHERE id = {user_id}").fetchall()

设置合理的内存上限： 默认 DuckDB 使用 80% 物理内存，在共享环境（CI、容器）中应显式限制。

con = duckdb.connect(config={
    "memory_limit": "4GB",
    "temp_directory": "/tmp/duckdb_spill",  # 超出内存时溢写磁盘
    "threads": 4,
})

大批量插入用 INSERT INTO ... SELECT 而非 executemany： executemany 每行单独执行，百万行级别比 bulk insert 慢 100 倍以上。

# 推荐：从 DataFrame 直接写入
duckdb.sql("INSERT INTO target_table SELECT * FROM df")

# 或注册后批量写
con.register("staging", df)
con.execute("INSERT INTO target_table SELECT * FROM staging")

# 不推荐：executemany 逐行插入
con.executemany("INSERT INTO target_table VALUES (?, ?, ?)", rows)

常见陷阱

陷阱：持久化数据库被多进程写锁阻塞

现象： 第二个进程打开同一个 .db 文件时抛出 IOException: Could not set lock on file，或程序卡住等待锁释放。

原因： DuckDB 持久化文件不支持多进程并发写入，同一时刻只能有一个写入连接。

解决： 多进程场景使用 read_only=True 只读连接（允许并发）；写入操作集中在单一进程；或改用内存数据库并定期 checkpoint。

# 读进程：只读模式，允许并发
reader = duckdb.connect("analytics.db", read_only=True)
df = reader.sql("SELECT * FROM monthly_kpi").to_df()

# 写进程：独占写连接，完成后务必关闭
with duckdb.connect("analytics.db") as writer:
    writer.execute("INSERT INTO ...")

陷阱：read_csv 类型推断错误导致数据丢失

现象： 某列被推断为 INTEGER 或 FLOAT，导致含字母的字符串值静默变为 NULL，或数值精度丢失。

原因： auto_detect=true 仅对前 sample_size（默认 20480）行采样，若前几万行全是数字、后面才出现字母，会推断错误类型。

解决： 显式指定有疑问的列类型，或加大 sample_size，或使用 all_varchar=true 先全读为字符串再转换。

# 错误：依赖自动推断
duckdb.sql("SELECT * FROM read_csv('data.csv', header=true)")

# 正确：显式指定关键列类型
duckdb.sql("""
    SELECT * FROM read_csv('data.csv',
        header=true,
        columns={'id': 'VARCHAR', 'code': 'VARCHAR', 'amount': 'DECIMAL(18,4)'}
    )
""")

# 或加大采样范围
duckdb.sql("""
    SELECT * FROM read_csv('data.csv', header=true, sample_size=-1)
""")  # -1 表示全量采样

陷阱：Relation 对象被垃圾回收后查询失败

现象： 将 from_df(df) 创建的 Relation 保存到变量，之后再查询时报 InvalidInputException 或结果为空。

原因： Relation 直接引用底层 Python 对象的内存（零拷贝）。如果原始 DataFrame df 被垃圾回收或被重新赋值，Relation 会访问已释放的内存。

解决： 确保原始对象的生命周期覆盖 Relation 的使用期，或在创建 Relation 后立即物化（调用 .to_df() 或 CREATE TABLE AS SELECT）。

def get_analysis():
    df = pd.read_parquet("data.parquet")
    rel = duckdb.from_df(df)
    # 错误：df 在函数返回后会被 GC，rel 持有悬空引用
    return rel  # 调用方使用 rel 时可能崩溃

# 正确：在函数内物化
def get_analysis():
    df = pd.read_parquet("data.parquet")
    return duckdb.from_df(df).to_df()  # 立即物化为新 DataFrame

陷阱：模块级默认连接在多线程中共享状态

现象： 多线程并发调用 duckdb.sql() 时出现查询结果混乱或报错。

原因： duckdb.sql() 使用全局默认连接，该连接不是线程安全的。

解决： 每个线程创建独立连接。

import threading
import duckdb

def worker(data):
    # 正确：每个线程独立连接
    with duckdb.connect() as con:
        return con.sql(f"SELECT count(*) FROM data").fetchone()[0]

# 错误：多线程共享 duckdb.sql（默认连接）
threads = [threading.Thread(target=lambda: duckdb.sql("SELECT ...")) for _ in range(4)]

参见

• SQLite完全指南 — 同为嵌入式数据库，适合 OLTP 场景
• ClickHouse完全指南 — 服务端列式 OLAP，适合 TB 级以上数据
• Pandas完全指南 — DuckDB 常见协作对象，零拷贝互操作
• Polars完全指南 — 高性能 DataFrame，与 DuckDB 互操作性同样优秀
• 数据库设计规范 — 建表和索引设计原则