duckdb

DuckDB——科学数据的 SQL 引擎

DuckDB 将专业 SQL 的强大功能带入 Python 数据科学技术栈。它针对“在线分析处理”（OLAP）进行了优化，擅长对规模大于内存容量的数据集进行大规模聚合、联接和复杂查询。

适用场景

• 对 Pandas 或 Polars 数据执行复杂的 SQL 查询（JOIN、窗口函数）。
• 直接查询大型 Parquet 或 CSV 文件，无需将其加载到内存中。
• 高效联接来自不同来源的数据（例如，一个 CSV 文件和一个 Pandas DataFrame）。
• 构建分析工作流，在这些工作流中，SQL 比 DataFrame 代码更简洁或更快。
• 管理规模超出 Excel 处理能力、但无需完整 PostgreSQL 服务的数据集。
• 用于机器学习的中间数据存储和特征工程。

参考文档

官方文档：https://duckdb.org/docs/
Python API：https://duckdb.org/docs/api/python/overview
搜索模式：duckdb.sql、duckdb.query、duckdb.read_parquet、duckdb.from_df

核心原则

进程内执行

DuckDB 在你的 Python 进程中运行，无需启动或管理服务。数据可以存储在文件（.db）中，也可以完全保存在内存中。

列式引擎

与 Polars 一样，DuckDB 使用列式存储和向量化执行引擎，在分析任务中，其速度比基于行的数据库（如 SQLite）快几个数量级。

无缝互操作

DuckDB 可以“看到”你的 Python 变量。你可以直接对 Pandas DataFrame 变量运行 SQL 查询，就像它是数据库中的一张表一样。

快速参考

安装

pip install duckdb

标准导入

import duckdb
import pandas as pd
import numpy as np

基本模式——查询 Python 数据

import duckdb
import pandas as pd

# 1. Create a sample DataFrame
df = pd.DataFrame({"id": [1, 2, 3], "val": [10.5, 20.0, 15.2]})

# 2. Query the DataFrame directly via SQL
# DuckDB automatically finds the 'df' variable in the local scope
result_df = duckdb.sql("SELECT id, val * 2 AS doubled FROM df WHERE val > 12").df()

print(result_df)

关键规则

✅ 应该做

• 直接查询文件——使用 SELECT * FROM 'data.parquet'，而不是先加载文件。DuckDB 只会读取所需的列和行。
• 使用 .df()、.pl()、.arrow() 方法——高效地将查询结果转换为你偏好的格式（Pandas、Polars 或 Arrow）。
• 大型数据使用持久化存储——如果希望数据在多次脚本运行之间保留，请使用 duckdb.connect('my_data.db')。
• 利用 Parquet——DuckDB 是处理 Parquet 文件的“同类最佳”引擎；使用 Parquet 可获得最高速度。
• 使用通配符——通过 FROM 'data/*.parquet' 一次查询数千个文件。
• 使用 EXPLAIN——在查询前加上 EXPLAIN ANALYZE，查看 DuckDB 如何执行查询并找出瓶颈。

❌ 不要做

• 不要用于高频写入（OLTP）——DuckDB 用于分析。如果你需要每秒数千次逐行插入，请使用 SQLite 或 PostgreSQL。
• 不要忽略连接——如果使用基于文件的数据库，请确保关闭连接或使用上下文管理器，以避免文件锁定。
• 不需要时不要手动加载 CSV——不要先执行 pd.read_csv() 再查询。直接查询文件路径可获得更好的性能。

反模式（绝对不要）

import duckdb
import pandas as pd

# ❌ BAD: Loading everything into Pandas just to do a simple filter
# df = pd.read_csv("massive.csv") 
# result = df[df['val'] > 100]

# ✅ GOOD: Let DuckDB filter while reading (saves RAM)
result = duckdb.sql("SELECT * FROM 'massive.csv' WHERE val > 100").df()

# ❌ BAD: Manual string formatting for SQL queries (SQL Injection risk)
# duckdb.sql(f"SELECT * FROM data WHERE name = '{user_input}'")

# ✅ GOOD: Use prepared statements or parameters
duckdb.execute("SELECT * FROM data WHERE name = ?", [user_input]).df()

# ❌ BAD: Re-reading the same file in a loop
# for i in range(10): 
#    res = duckdb.sql("SELECT mean(val) FROM 'large.parquet'").df()

# ✅ GOOD: Create a VIEW or TABLE first
duckdb.sql("CREATE VIEW data_view AS SELECT * FROM 'large.parquet'")
for i in range(10):
    res = duckdb.sql("SELECT mean(val) FROM data_view").df()

SQL 功能和操作

查询不同来源

# Query a CSV
res_csv = duckdb.sql("SELECT * FROM read_csv_auto('data.csv')").df()

# Query a Parquet file
res_pq = duckdb.sql("SELECT * FROM 'data.parquet' WHERE price > 50").df()

# Query multiple Parquet files and join with a Pandas DF
df_metadata = pd.DataFrame(...)
query = """
    SELECT p.*, m.category 
    FROM 'raw_data/*.parquet' p
    JOIN df_metadata m ON p.id = m.id
    LIMIT 10
"""
res = duckdb.sql(query).df()

关系 API（程序化 SQL）

如果你更喜欢 Python 风格的链式方法，而不是原生 SQL：

rel = duckdb.from_df(df)
res = rel.filter("val > 15").project("id, val * 2").order("val").limit(5)
print(res.df())

高级 SQL：窗口函数和聚合

DuckDB 支持完整的现代 SQL；在处理复杂统计时，它通常比 Pandas 更简单。

query = """
    SELECT 
        date,
        station_id,
        temp,
        AVG(temp) OVER (PARTITION BY station_id ORDER BY date ROWS BETWEEN 7 PRECEDING AND CURRENT ROW) as rolling_7d_avg,
        temp - LAG(temp) OVER (PARTITION BY station_id ORDER BY date) as daily_change
    FROM 'weather_data.parquet'
"""
df_stats = duckdb.sql(query).df()

使用持久化数据库

# Create or open a database file
con = duckdb.connect('scientific_project.db')

# Create a table from a dataframe
con.execute("CREATE TABLE experiment_results AS SELECT * FROM df")

# Check tables
print(con.execute("SHOW TABLES").df())

# Close connection
con.close()

性能优化

1. 外部聚合（溢写到磁盘）

如果查询超出 RAM 容量，DuckDB 可以“溢写到磁盘”以完成计算。

# Enable temp directory for large queries
duckdb.sql("SET temp_directory='/tmp/duckdb_temp/'")
duckdb.sql("SET max_memory='4GB'") # Limit RAM usage

2. 并行处理

DuckDB 默认使用并行处理。你可以控制线程数。

duckdb.sql("SET threads TO 8")

3. 探索性分析中的抽样

查询大型文件中的样本可以瞬间完成。

# Random 10% sample
df_sample = duckdb.sql("SELECT * FROM 'huge.parquet' USING SAMPLE 10%").df()

实用工作流

1. “大数据”清理工作流

def process_experiment_logs(glob_pattern):
    """Clean and aggregate TBs of log data across many files."""
    query = f"""
    WITH clean_data AS (
        SELECT 
            timestamp::TIMESTAMP as ts,
            sensor_id,
            value
        FROM read_csv_auto('{glob_pattern}')
        WHERE value IS NOT NULL AND value != -999
    )
    SELECT 
        time_bucket(INTERVAL '1 hour', ts) as hour,
        sensor_id,
        AVG(value) as avg_val
    FROM clean_data
    GROUP BY 1, 2
    ORDER BY 1, 2
    """
    return duckdb.sql(query).df()

2. 面向 ML 的快速特征工程

def create_features(df_train):
    # Use SQL to create complex lag and moving average features
    return duckdb.sql("""
        SELECT *,
               AVG(price) OVER (PARTITION BY item ORDER BY date ROWS 3 PRECEDING) as ma3,
               COUNT(*) OVER (PARTITION BY user) as user_activity_count
        FROM df_train
    """).df()

3. 互操作：DuckDB 到 PyTorch/TensorFlow

# Query data and convert to Arrow for zero-copy transfer to Deep Learning
arrow_table = duckdb.sql("SELECT * FROM 'data.parquet'").arrow()

# Then in PyTorch (requires torch.utils.dlpack or similar)
# Or just use the fast arrow-to-numpy/tensor path

常见问题与解决方案

“未找到变量”错误

DuckDB 会在调用 duckdb.sql() 的作用域中查找变量。

# ✅ Solution: If calling inside a function, ensure the variable is local or passed
def my_query(my_df):
    return duckdb.sql("SELECT * FROM my_df").df()

数据库文件被锁定

如果两个进程尝试以“写入”模式打开同一个 .db 文件，就会失败。

# ✅ Solution: Open as read-only for secondary processes
con = duckdb.connect('data.db', read_only=True)

日期/时间解析

CSV 中的日期格式往往很奇怪。

# ✅ Solution: Use strptime or let DuckDB auto-detect
# SELECT strptime(date_col, '%d/%m/%Y') FROM 'data.csv'

最佳实践

1. 直接查询文件 - 使用 SELECT * FROM 'file.parquet'，而不是先将文件加载到内存中。
2. 使用合适的输出格式 - Pandas 使用 .df()，Polars 使用 .pl()，Arrow 使用 .arrow()。
3. 为重复查询创建视图 - 避免多次重新读取同一个文件。
4. 使用预处理语句 - 防止 SQL 注入，并提升重复查询的性能。
5. 充分利用 Parquet 格式 - DuckDB 擅长处理 Parquet；使用它可获得最佳性能。
6. 使用通配符查询多个文件 - 使用 FROM 'data/*.parquet' 一次查询多个文件。
7. 为大型查询启用磁盘溢写 - 为内存不足时的操作设置 temp_directory 和 max_memory。
8. 使用 EXPLAIN ANALYZE - 了解查询的执行过程并找出性能瓶颈。
9. 正确关闭连接 - 使用上下文管理器，或显式关闭基于文件的连接。
10. 复杂聚合优先使用 SQL - 与 DataFrame 代码相比，使用 SQL 编写窗口函数和复杂联接通常更清晰。

DuckDB 是连接 SQL 分析能力与 Python 灵活性的桥梁。它免去了“数据加载成本”，让科学家能够专注于以极快的速度向数据提出复杂问题。