Python数据仓库：如何使用DuckDB和Polars库在Python中进行高性能OLAP查询。

Python数据仓库：DuckDB与Polars构建高性能OLAP查询

大家好，今天我们来聊聊如何利用Python构建高性能的数据仓库，重点在于使用 DuckDB 和 Polars 这两个强大的库进行 OLAP (Online Analytical Processing) 查询。

传统的Python在数据分析方面，由于其解释型语言的特性，在处理大规模数据时经常显得力不从心。Pandas虽然功能强大，但在内存限制和速度方面也存在瓶颈。DuckDB 和 Polars 的出现，为Python数据分析带来了新的可能，它们专注于高性能和易用性，使得我们可以在Python环境中高效地进行数据仓库级别的分析。

什么是OLAP？

在深入代码之前，我们先简单回顾一下OLAP的概念。OLAP 旨在快速响应多维数据分析请求。与 OLTP (Online Transaction Processing) 强调事务处理和数据更新不同，OLAP 侧重于数据的查询和分析，常用于决策支持系统。OLAP 操作通常包括：

• 切片 (Slice): 从一个维度中选择一个特定的值，从而减少数据的维度。
• 切块 (Dice): 从多个维度中选择多个特定的值，进一步限制数据集。
• 钻取 (Drill-down): 从汇总级别向下查看更详细的数据。
• 上卷 (Roll-up): 从详细级别向上查看汇总的数据。
• 透视 (Pivot): 旋转数据的维度，以不同的角度查看数据。

DuckDB：嵌入式分析型数据库

DuckDB 是一个开源的、高性能的嵌入式分析型数据库。它的主要特点包括：

• 嵌入式： DuckDB 运行在进程内，不需要单独的服务器，方便部署和使用。
• AP 型数据库： 针对分析查询进行了优化，性能优异。
• 支持 SQL： 完全支持 SQL 标准，方便用户进行数据查询。
• Zero-copy Pandas 集成： 可以直接从 Pandas DataFrame 读取数据，避免数据复制，提高性能。
• 支持多种数据格式： 可以直接读取 CSV、Parquet 等文件格式。

DuckDB 的安装与基本使用

# 安装 DuckDB
# pip install duckdb

import duckdb

# 创建一个 DuckDB 数据库连接 (内存数据库)
con = duckdb.connect(':memory:')

# 创建一个表
con.execute("""
CREATE TABLE sales (
    order_id INTEGER,
    customer_id INTEGER,
    product_id INTEGER,
    order_date DATE,
    quantity INTEGER,
    price REAL
);
""")

# 插入一些数据
con.execute("""
INSERT INTO sales VALUES
(1, 101, 201, '2023-01-01', 2, 10.0),
(2, 102, 202, '2023-01-01', 1, 20.0),
(3, 101, 203, '2023-01-02', 3, 5.0),
(4, 103, 201, '2023-01-02', 1, 10.0),
(5, 102, 202, '2023-01-03', 2, 20.0);
""")

# 执行一个查询
result = con.execute("SELECT customer_id, SUM(quantity * price) AS total_sales FROM sales GROUP BY customer_id ORDER BY total_sales DESC").fetchall()

print(result)  # 输出: [(102, 60.0), (101, 35.0), (103, 10.0)]

# 关闭连接
con.close()

这段代码演示了如何使用 DuckDB 创建一个简单的 sales 表，插入一些数据，并执行一个聚合查询。 con.connect(':memory:') 创建了一个内存数据库，这意味着数据不会持久化到磁盘。你可以将 :memory: 替换为文件路径，例如 con.connect('sales.duckdb')，以创建一个持久化的数据库。

从 Pandas DataFrame 读取数据

DuckDB 提供了无缝的 Pandas 集成，可以直接从 Pandas DataFrame 读取数据，而无需进行数据复制。

import pandas as pd
import duckdb

# 创建一个 Pandas DataFrame
data = {'customer_id': [101, 102, 101, 103, 102],
        'product_id': [201, 202, 203, 201, 202],
        'quantity': [2, 1, 3, 1, 2],
        'price': [10.0, 20.0, 5.0, 10.0, 20.0]}
df = pd.DataFrame(data)

# 连接到 DuckDB
con = duckdb.connect(':memory:')

# 将 DataFrame 注册为 DuckDB 的一个视图
con.register('sales_df', df)

# 执行 SQL 查询
result = con.execute("SELECT customer_id, SUM(quantity * price) AS total_sales FROM sales_df GROUP BY customer_id ORDER BY total_sales DESC").df()

print(result)

# 关闭连接
con.close()

con.register('sales_df', df) 这行代码将 Pandas DataFrame df 注册为 DuckDB 的一个视图，名为 sales_df。 之后，你就可以像查询普通表一样查询这个视图了。 con.execute(...).df() 这行代码将查询结果转换为 Pandas DataFrame，方便后续处理。这种方式避免了数据的复制，提高了性能。

从 CSV 文件读取数据

DuckDB 可以直接从 CSV 文件读取数据，而不需要先将数据加载到 Pandas DataFrame 中。

import duckdb

# 创建一个 CSV 文件 (假设文件名为 sales.csv)
# 文件内容:
# customer_id,product_id,quantity,price
# 101,201,2,10.0
# 102,202,1,20.0
# 101,203,3,5.0
# 103,201,1,10.0
# 102,202,2,20.0

# 连接到 DuckDB
con = duckdb.connect(':memory:')

# 从 CSV 文件创建表
con.execute("""
CREATE TABLE sales AS
SELECT * FROM read_csv_auto('sales.csv');
""")

# 执行 SQL 查询
result = con.execute("SELECT customer_id, SUM(quantity * price) AS total_sales FROM sales GROUP BY customer_id ORDER BY total_sales DESC").fetchall()

print(result)

# 关闭连接
con.close()

read_csv_auto('sales.csv') 这个函数会自动检测 CSV 文件的分隔符、数据类型等，并将其转换为 DuckDB 的表。 这种方式非常方便，可以快速地将 CSV 文件加载到 DuckDB 中进行分析。

DuckDB 的高级功能

DuckDB 还支持许多高级功能，例如：

• Parquet 支持： 可以直接读取和写入 Parquet 文件，这是一种列式存储格式，非常适合 OLAP 查询。
• JSON 支持： 可以直接查询 JSON 数据。
• UDF (User-Defined Functions)： 可以定义自己的函数，并在 SQL 查询中使用。

我们将在后面的例子中看到 Parquet 的使用。

Polars：高性能 DataFrame 库

Polars 是一个用 Rust 编写的 DataFrame 库，旨在提供极致的性能和易用性。它的主要特点包括：

• 基于 Arrow： Polars 基于 Apache Arrow 构建，这是一种内存中的列式数据格式，可以实现零拷贝的数据传输。
• 并行处理： Polars 充分利用多核 CPU 进行并行处理，提高查询速度。
• 延迟执行： Polars 使用查询优化器，将查询计划优化后再执行，从而提高性能。
• 内存效率： Polars 在内存使用方面进行了优化，可以处理比 Pandas 更大的数据集。
• 支持多种数据格式： 可以读取 CSV、Parquet、JSON 等文件格式。

Polars 的安装与基本使用

# 安装 Polars
# pip install polars

import polars as pl

# 创建一个 Polars DataFrame
data = {'customer_id': [101, 102, 101, 103, 102],
        'product_id': [201, 202, 203, 201, 202],
        'quantity': [2, 1, 3, 1, 2],
        'price': [10.0, 20.0, 5.0, 10.0, 20.0]}
df = pl.DataFrame(data)

# 执行一个查询
result = df.group_by('customer_id').agg([
    pl.col('quantity').sum().alias('total_quantity'),
    pl.col('price').mean().alias('average_price')
]).sort('total_quantity', descending=True)

print(result)

这段代码演示了如何使用 Polars 创建一个 DataFrame，并执行一个分组聚合查询。 pl.DataFrame(data) 创建了一个 Polars DataFrame。 df.group_by('customer_id') 按照 customer_id 进行分组。 pl.col('quantity').sum().alias('total_quantity') 计算每个分组的 quantity 的总和，并将结果命名为 total_quantity。 pl.col('price').mean().alias('average_price') 计算每个分组的 price 的平均值，并将结果命名为 average_price。 sort('total_quantity', descending=True) 按照 total_quantity 降序排序。

Polars 的延迟执行

Polars 采用延迟执行的策略，这意味着查询计划不会立即执行，而是会被优化器进行优化后再执行。 这种策略可以提高查询性能，尤其是在复杂的查询中。

import polars as pl

# 创建一个 Polars DataFrame
data = {'customer_id': [101, 102, 101, 103, 102],
        'product_id': [201, 202, 203, 201, 202],
        'quantity': [2, 1, 3, 1, 2],
        'price': [10.0, 20.0, 5.0, 10.0, 20.0]}
df = pl.DataFrame(data)

# 定义一个查询表达式
query = df.group_by('customer_id').agg([
    pl.col('quantity').sum().alias('total_quantity'),
    pl.col('price').mean().alias('average_price')
]).sort('total_quantity', descending=True)

# 查看查询计划
print(query.explain())

# 执行查询
result = query.collect()

print(result)

query.explain() 可以查看 Polars 的查询计划。 query.collect() 强制执行查询。 通过查看查询计划，你可以了解 Polars 是如何优化你的查询的。

Polars 的高性能

Polars 的高性能主要来自于以下几个方面：

• 基于 Arrow： Arrow 提供了零拷贝的数据传输，避免了数据复制的开销。
• 并行处理： Polars 充分利用多核 CPU 进行并行处理。
• 查询优化器： Polars 使用查询优化器，将查询计划优化后再执行。
• Rust 编写： Rust 是一种高性能的系统编程语言，可以提供接近 C++ 的性能。

Polars 的高级功能

Polars 还支持许多高级功能，例如：

• 窗口函数： 可以执行窗口函数操作，例如计算移动平均值。
• 连接操作： 支持多种连接操作，例如 inner join、left join、right join、full outer join。
• 表达式语言： 提供强大的表达式语言，可以进行复杂的计算和转换。

DuckDB + Polars：强强联合

DuckDB 和 Polars 可以很好地结合使用，发挥各自的优势。 你可以使用 DuckDB 加载和处理数据，然后将数据传递给 Polars 进行进一步的分析。

import duckdb
import polars as pl

# 创建一个 DuckDB 数据库连接
con = duckdb.connect(':memory:')

# 创建一个表
con.execute("""
CREATE TABLE sales (
    order_id INTEGER,
    customer_id INTEGER,
    product_id INTEGER,
    order_date DATE,
    quantity INTEGER,
    price REAL
);
""")

# 插入一些数据
con.execute("""
INSERT INTO sales VALUES
(1, 101, 201, '2023-01-01', 2, 10.0),
(2, 102, 202, '2023-01-01', 1, 20.0),
(3, 101, 203, '2023-01-02', 3, 5.0),
(4, 103, 201, '2023-01-02', 1, 10.0),
(5, 102, 202, '2023-01-03', 2, 20.0);
""")

# 从 DuckDB 读取数据到 Polars DataFrame
df = pl.from_arrow(con.execute("SELECT * FROM sales").fetch_arrow_table())

# 使用 Polars 进行分析
result = df.group_by('customer_id').agg([
    pl.col('quantity').sum().alias('total_quantity'),
    pl.col('price').mean().alias('average_price')
]).sort('total_quantity', descending=True)

print(result)

# 关闭连接
con.close()

pl.from_arrow(con.execute("SELECT * FROM sales").fetch_arrow_table()) 这行代码将 DuckDB 的查询结果转换为 Apache Arrow 格式，然后使用 Polars 从 Arrow 格式创建 DataFrame。 这种方式可以避免数据的复制，提高性能。

案例：分析大型 Parquet 数据集

现在，我们来看一个更复杂的例子，演示如何使用 DuckDB 和 Polars 分析大型 Parquet 数据集。 假设我们有一个包含订单数据的 Parquet 文件，文件名为 orders.parquet。 这个文件包含以下列：

列名	数据类型	描述
order_id	INTEGER	订单 ID
customer_id	INTEGER	客户 ID
product_id	INTEGER	产品 ID
order_date	DATE	订单日期
quantity	INTEGER	订单数量
price	REAL	产品价格

我们的目标是计算每个客户的总消费金额，并找出消费金额最高的 10 个客户。

import duckdb
import polars as pl

# 连接到 DuckDB
con = duckdb.connect(':memory:')

# 从 Parquet 文件创建表
con.execute("""
CREATE TABLE orders AS
SELECT * FROM read_parquet('orders.parquet');
""")

# 将数据从 DuckDB 传递给 Polars
df = pl.from_arrow(con.execute("SELECT * FROM orders").fetch_arrow_table())

# 使用 Polars 进行分析
result = (
    df.group_by('customer_id')
    .agg([
        (pl.col('quantity') * pl.col('price')).sum().alias('total_spending')
    ])
    .sort('total_spending', descending=True)
    .head(10)
)

print(result)

# 关闭连接
con.close()

在这个例子中，我们首先使用 DuckDB 从 Parquet 文件创建表。 然后，我们将数据从 DuckDB 传递给 Polars。 最后，我们使用 Polars 进行分组聚合、排序和截取操作，计算出消费金额最高的 10 个客户。 这个例子展示了 DuckDB 和 Polars 的强大功能，以及它们在处理大型数据集时的性能优势。

一些优化建议

• 数据类型： 选择合适的数据类型可以减少内存使用和提高查询速度。例如，如果 customer_id 的取值范围较小，可以使用 INTEGER 而不是 BIGINT。
• 索引： 在经常用于查询的列上创建索引可以提高查询速度。DuckDB 支持多种索引类型，例如 B-tree 索引和哈希索引。
• 分区： 将大型数据集分成多个分区可以提高查询速度。DuckDB 和 Polars 都支持分区表。
• 查询优化： 编写高效的 SQL 查询可以提高查询速度。 例如，避免使用 SELECT *，而是只选择需要的列。
• 硬件： 使用更快的 CPU 和更大的内存可以提高整体性能。

总结

DuckDB 和 Polars 是两个强大的 Python 库，可以帮助你构建高性能的数据仓库，并进行高效的 OLAP 查询。 DuckDB 提供了嵌入式数据库的功能，可以方便地加载和处理数据。 Polars 提供了高性能的 DataFrame 操作，可以进行复杂的分析。 通过将 DuckDB 和 Polars 结合使用，你可以充分利用它们的优势，构建出满足你需求的分析系统。

持续学习

希望今天的讲解能够帮助大家更好地理解 DuckDB 和 Polars 在 Python 数据仓库中的应用。这两个库都在不断发展，持续关注它们的更新和文档，可以帮助你更好地利用它们的功能。 实践是最好的老师，尝试使用 DuckDB 和 Polars 处理你自己的数据，你将会发现它们在数据分析方面的巨大潜力。