PostgreSQL 窗口函数实战：从移动平均到排名，结合 Pandas 实现高效数据分析

你好！相信你作为一名开发者或者数据科学家，一定经常遇到需要进行复杂数据分析的场景。比如，计算移动平均值、对数据进行排名、计算百分位数等等。虽然 Pandas 提供了强大的数据处理能力，但在处理海量数据时，直接在数据库层面进行预处理往往更加高效。今天，咱们就来聊聊 PostgreSQL 的窗口函数，以及如何将它与 Pandas 结合，实现更高效的数据分析。

什么是窗口函数？

先别被“窗口”这个词吓到，它其实很简单。你可以把它想象成一个“滑动窗口”，这个窗口在你的数据表上滑动，每次只处理窗口内的数据。与 GROUP BY 不同，窗口函数不会将数据分组，而是对每一行数据都应用计算，并将结果添加到该行。

举个例子，假设你有一个销售数据表 sales，包含 product (产品)、sale_date (销售日期) 和 amount (销售额) 三列。你想计算每个产品每天的销售额以及过去 7 天的移动平均销售额。使用窗口函数，你可以轻松实现这个需求，而无需编写复杂的 SQL 查询或使用循环。

窗口函数的基本语法

PostgreSQL 窗口函数的基本语法如下：

<窗口函数> OVER (
    [PARTITION BY <分组列>]
    [ORDER BY <排序列>]
    [ROWS BETWEEN <起始行> AND <结束行>]
)

• <窗口函数>：你要使用的窗口函数，例如 AVG (平均值)、SUM (求和)、RANK (排名)、ROW_NUMBER (行号) 等。
• PARTITION BY：可选，指定分组列。类似于 GROUP BY，但窗口函数不会将数据分组，而是对每个分组内的每一行都应用计算。
• ORDER BY：可选，指定排序列。用于确定窗口内数据的顺序。
• ROWS BETWEEN ... AND ...：可选，指定窗口的范围。可以指定相对当前行的偏移量，例如 ROWS BETWEEN 7 PRECEDING AND CURRENT ROW 表示当前行及之前的 7 行。

常用窗口函数

PostgreSQL 提供了丰富的窗口函数，咱们挑几个常用的来详细说说。

1. ROW_NUMBER()：行号

ROW_NUMBER() 函数为每一行分配一个唯一的序号，从 1 开始递增。即使有相同的值，也会分配不同的序号。

SELECT
    product,
    sale_date,
    amount,
    ROW_NUMBER() OVER (ORDER BY sale_date) AS row_num
FROM sales;

这个查询会按照 sale_date 对 sales 表进行排序，并为每一行分配一个行号。

2. RANK() 和 DENSE_RANK()：排名

RANK() 和 DENSE_RANK() 函数都用于排名，但它们处理相同值的方式不同。

• RANK()：允许并列排名，并跳过后续排名。例如，如果有两个并列第一，则下一个排名是第三。
• DENSE_RANK()：允许并列排名，但不跳过后续排名。例如，如果有两个并列第一，则下一个排名是第二。

SELECT
    product,
    sale_date,
    amount,
    RANK() OVER (PARTITION BY product ORDER BY amount DESC) AS rank,
    DENSE_RANK() OVER (PARTITION BY product ORDER BY amount DESC) AS dense_rank
FROM sales;

这个查询会按照 product 分组，并在每个分组内按照 amount 降序排名。rank 列会跳过排名，dense_rank 列不会跳过排名。

3. LAG() 和 LEAD()：获取前后行的值

LAG() 和 LEAD() 函数可以获取当前行之前或之后的行的值。

• LAG(column, offset, default)：获取当前行之前 offset 行的 column 列的值。如果不存在，则返回 default 值（默认为 NULL）。
• LEAD(column, offset, default)：获取当前行之后 offset 行的 column 列的值。如果不存在，则返回 default 值（默认为 NULL）。

SELECT
    product,
    sale_date,
    amount,
    LAG(amount, 1, 0) OVER (PARTITION BY product ORDER BY sale_date) AS previous_day_amount,
    LEAD(amount, 1, 0) OVER (PARTITION BY product ORDER BY sale_date) AS next_day_amount
FROM sales;

这个查询会按照 product 分组，并在每个分组内按照 sale_date 排序。previous_day_amount 列会获取前一天的销售额，next_day_amount 列会获取后一天的销售额。

4. AVG()、SUM()、MIN()、MAX()：聚合函数

你可以在窗口函数中使用聚合函数，例如 AVG、SUM、MIN、MAX 等。这可以让你计算移动平均值、累计和等。

SELECT
    product,
    sale_date,
    amount,
    AVG(amount) OVER (PARTITION BY product ORDER BY sale_date ROWS BETWEEN 7 PRECEDING AND CURRENT ROW) AS moving_average
FROM sales;

这个查询会按照 product 分组，并在每个分组内按照 sale_date 排序。moving_average 列会计算当前行及之前 7 天的平均销售额。

结合 Pandas 实现高效数据分析

PostgreSQL 窗口函数可以进行强大的数据预处理，但 Pandas 在数据可视化、数据探索和更复杂的数据分析方面更胜一筹。因此，将两者结合起来可以实现更高效的数据分析。

你可以使用 psycopg2 或 SQLAlchemy 等库连接 PostgreSQL 数据库，并将查询结果读取到 Pandas DataFrame 中。

import pandas as pd
import psycopg2
 
# 连接数据库
conn = psycopg2.connect(database="your_database", user="your_user", password="your_password", host="your_host", port="your_port")
 
# 执行查询
query = """
SELECT
    product,
    sale_date,
    amount,
    AVG(amount) OVER (PARTITION BY product ORDER BY sale_date ROWS BETWEEN 7 PRECEDING AND CURRENT ROW) AS moving_average
FROM sales;
"""
df = pd.read_sql_query(query, conn)
 
# 关闭连接
conn.close()
 
# 使用 Pandas 进行进一步分析
print(df.head())
# ...

将数据读取到 Pandas DataFrame 后，你可以使用 Pandas 的各种函数进行进一步分析和可视化。

实际案例：电商销售数据分析

假设你是一家电商公司的数据分析师，你需要分析过去一年每个产品的销售情况，并计算以下指标：

• 每个产品每天的销售额
• 每个产品过去 7 天的移动平均销售额
• 每个产品每天的销售额排名
• 每个产品累计销售额

你可以使用以下 SQL 查询和 Pandas 代码实现这个需求：

-- 创建 sales 表（如果不存在）
CREATE TABLE IF NOT EXISTS sales (
    product VARCHAR(255),
    sale_date DATE,
    amount DECIMAL(10, 2)
);
 
-- 插入示例数据（如果表为空）
INSERT INTO sales (product, sale_date, amount)
SELECT
    'Product ' || (floor(random() * 5) + 1)::text,
    CURRENT_DATE - (floor(random() * 365)) * INTERVAL '1 day',
    (random() * 1000)::numeric(10, 2)
FROM generate_series(1, 1000);
 
-- 查询数据并计算指标
SELECT
    product,
    sale_date,
    amount,
    AVG(amount) OVER (PARTITION BY product ORDER BY sale_date ROWS BETWEEN 6 PRECEDING AND CURRENT ROW) AS moving_average,
    RANK() OVER (PARTITION BY product ORDER BY amount DESC) AS daily_rank,
    SUM(amount) OVER (PARTITION BY product ORDER BY sale_date) AS cumulative_sales
FROM sales;

import pandas as pd
import psycopg2
 
# 连接数据库（替换为你的数据库信息）
conn = psycopg2.connect(database="your_database", user="your_user", password="your_password", host="your_host", port="your_port")
 
# 执行查询
query = """
SELECT
    product,
    sale_date,
    amount,
    AVG(amount) OVER (PARTITION BY product ORDER BY sale_date ROWS BETWEEN 6 PRECEDING AND CURRENT ROW) AS moving_average,
    RANK() OVER (PARTITION BY product ORDER BY amount DESC) AS daily_rank,
    SUM(amount) OVER (PARTITION BY product ORDER BY sale_date) AS cumulative_sales
FROM sales;
"""
df = pd.read_sql_query(query, conn)
 
# 关闭连接
conn.close()
 
# 使用 Pandas 进行进一步分析和可视化
print(df.head())
 
# 例如，绘制每个产品的移动平均销售额
import matplotlib.pyplot as plt
 
for product in df['product'].unique():
    product_df = df[df['product'] == product]
    plt.plot(product_df['sale_date'], product_df['moving_average'], label=product)
 
plt.xlabel('Sale Date')
plt.ylabel('Moving Average')
plt.title('7-Day Moving Average Sales by Product')
plt.legend()
plt.show()
 
# ... 其他分析和可视化 ...

这个案例展示了如何使用 PostgreSQL 窗口函数和 Pandas 结合进行实际的数据分析。你可以根据自己的需求修改查询和代码。

总结

PostgreSQL 窗口函数是一个强大的工具，可以让你在数据库层面进行复杂的数据分析。结合 Pandas，你可以实现更高效的数据处理和可视化。希望这篇文章能帮助你更好地理解和使用 PostgreSQL 窗口函数！如果你有任何问题或想法，欢迎在评论区留言。

总而言之，窗口函数就像一把瑞士军刀，能帮你处理各种数据分析难题。掌握它，你的数据分析能力一定会更上一层楼！