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 处理你自己的数据，你将会发现它们在数据分析方面的巨大潜力。