DuckDB实践教程：Python数据分析新利器

一、DuckDB概览

1.1 什么是DuckDB？

DuckDB是一款面向分析型工作负载的开源嵌入式数据库，具有以下特性：

内存计算：数据可完全存储在内存中处理

SQL支持：完整实现ANSI SQL标准

Python集成：通过duckdb模块直接操作

高性能：采用向量化查询执行引擎

1.2 安装与基础操作

# 安装命令（PEP8规范）

import duckdb

# 创建内存数据库

con = duckdb.connect(database=':memory:')

# 执行基础查询

result = con.execute("SELECT 'Hello DuckDB'").fetchall()

print(result)  # [('Hello DuckDB',)]

二、核心功能解析

2.1 数据导入导出函数原型：

# 从Pandas DataFrame导入

con.register('table_name', dataframe)

# 导出到Pandas DataFrame

df = con.execute("SELECT * FROM table").fetchdf()

案例：

import pandas as pd

# 创建示例数据

sales_data = pd.DataFrame({

'date': ['2023-01-01', '2023-01-02'],

'amount': [1500, 2300]

})

# 注册DataFrame

con.register('sales', sales_data)

# 执行复杂查询

query = """

SELECT

  SUM(amount) AS total,

  AVG(amount) AS average

FROM sales

"""

result = con.execute(query).fetchdf()

print(result)

2.2 查询技巧

参数说明：

# 参数化查询（防止SQL注入）

con.execute("SELECT * FROM tbl WHERE id=?", [1001])

# 结果分页

con.execute("SELECT * FROM users LIMIT 10 OFFSET 20")

性能对比：

2.3 高级分析功能

窗口函数示例：

query = """

SELECT

  employee_id,

  sales,

  RANK() OVER (ORDER BY sales DESC) AS rank

FROM sales_records

"""

con.execute(query)

时间序列处理：

query = """

SELECT

  DATE_TRUNC('month', order_date) AS month,

  SUM(amount) AS monthly_sales

FROM orders

GROUP BY month

ORDER BY month

"""

三、实战应用案例

3.1 数据分析流水线

实现代码：

def data_pipeline(csv_path):

# 创建持久化数据库

  con = duckdb.connect('analysis.db')

# 直接查询CSV文件

  con.execute(f"""

      CREATE TABLE sales AS

      SELECT * FROM read_csv_auto('{csv_path}')

  """)

# 数据清洗

  con.execute("""

      DELETE FROM sales

      WHERE amount < 0 OR quantity <= 0

  """)

# 生成报告

  report = con.execute("""

      SELECT

          product_category,

          SUM(amount) AS total_sales,

          COUNT(*) AS transactions

      FROM sales

      GROUP BY product_category

  """).fetchdf()

return report

3.2 实时数据处理

from queue import Queue

classRealTimeProcessor:

def__init__(self):

self.con = duckdb.connect()

self.data_queue = Queue()

defingest_data(self, record):

"""数据注入方法"""

self.data_queue.put(record)

defprocess_stream(self):

"""流式处理核心"""

whileTrue:

ifnotself.data_queue.empty():

              batch = [self.data_queue.get() for _ inrange(100)]

              df = pd.DataFrame(batch)

self.con.register('temp_batch', df)

self.con.execute("""

                  INSERT INTO main_table

                  SELECT * FROM temp_batch

              """)

# 执行实时分析

self.con.execute("""

              SELECT

                  COUNT(*) AS total,

                  AVG(value) AS average

              FROM main_table

          """)

四、优化建议

4.1 性能优化技巧

批量插入

：使用COPY命令代替多次INSERT

con.execute("COPY sales FROM 'data.csv' (HEADER)")

索引策略

：对常用过滤字段创建索引

con.execute("CREATE INDEX idx_user_id ON users(id)")

内存管理

：定期清理缓存

con.execute("PRAGMA memory_limit='1GB'")

4.2 常见问题排查

五、扩展应用

5.1 与机器学习整合

from sklearn.linear_model import LinearRegression

# 从数据库加载训练数据

train_data = con.execute("""

  SELECT feature1, feature2, target

  FROM training_data

""").fetchdf()

# 训练模型

model = LinearRegression()

model.fit(train_data[['feature1', 'feature2']], train_data['target'])

5.2 可视化集成

import matplotlib.pyplot as plt

# 执行聚合查询

sales_trend = con.execute("""

  SELECT month, SUM(sales)

  FROM monthly_sales

  GROUP BY month

""").fetchdf()

# 生成可视化

plt.figure(figsize=(10,6))

plt.plot(sales_trend['month'], sales_trend['sum(sales)'])

plt.title('Monthly Sales Trend')

plt.xlabel('Month')

plt.ylabel('Sales')

plt.grid(True)

plt.show()

六、总结与资源

学习路径：

官方文档：https://duckdb.org/docs/

示例仓库：https://github.com/duckdb/duckdb-python

性能白皮书：https://duckdb.org/why_duckdb.html

通过本教程，我们可以掌握DuckDB的核心操作和实战技巧。建议从简单的数据分析任务开始实践，逐步扩展到复杂场景。建议关注内存管理和索引策略，这对保持系统高性能比较重要。

更新日期：2025-05-26

交流讨论：欢迎在评论区留言！

重要提示：本文主要是记录自己的学习与实践过程，所提内容或者观点仅代表个人意见，只是我以为的，不代表完全正确，不喜请勿关注。