Spark SQL 窗口函数高级应用：优化查询性能与实现复杂数据分析

Spark SQL 窗口函数高级应用：优化查询性能与实现复杂数据分析

窗口函数是 Spark SQL 中强大的分析工具，能够在不改变数据行数的情况下，为每一行计算基于窗口内其他行的结果。相比传统聚合操作，窗口函数更灵活高效，尤其适合复杂业务分析场景。本文将深入探讨窗口函数的高级应用，包括累积聚合、分组 TopN 与复杂数据分析，帮助读者掌握优化查询性能与实现复杂数据分析的方法。

1. 窗口函数基础与概念

窗口函数是 SQL 中的一类特殊函数，它对一组行（窗口）执行计算，但不会将多行压缩成单行输出，这与传统聚合函数形成鲜明对比。窗口函数结合了分组和排序的特点，既能保持原始数据的行数，又能进行复杂计算。

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

函数名(列) OVER ([PARTITION BY 分组列] [ORDER BY 排序列] [窗口范围])

其中：

• PARTITION BY 定义分组的列，类似于 GROUP BY 但不会减少行数
• ORDER BY 定义排序的列，决定窗口内行的顺序
• 窗口范围 定义窗口的大小，如 ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW

窗口函数的种类包括：

相比传统聚合函数，窗口函数的主要优势在于：

• 不改变原始数据的行数
• 可以同时访问多个窗口范围
• 计算结果更加灵活，可以结合多列计算
• 执行效率更高，尤其在大数据场景下

2. 累积聚合应用

累积聚合是窗口函数的经典应用场景，可以计算分组内从开始到当前行的累积值。常见应用包括累积销售额、累积用户增长、累积订单量等。

累积聚合的核心语法：

SUM(列) OVER (PARTITION BY 分组列 ORDER BY 时间列
ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)

2.1 电商销售累积分析

假设我们有销售数据表，需要计算每个产品类别的累积销售额：

SELECT
product_category,
month,
sales_amount,
SUM(sales_amount) OVER (
PARTITION BY product_category
ORDER BY month
ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
) AS cumulative_sales
FROM sales_data
ORDER BY product_category, month;

这段代码中：

• PARTITION BY product_category 按产品类别分组
• ORDER BY month 按月份排序
• ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW 定义窗口从分组开始到当前行

2.2 用户增长趋势分析

在用户行为分析中，累积聚合可用于计算用户增长趋势：

SELECT
date,
new_users,
SUM(new_users) OVER (
ORDER BY date
ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
) AS cumulative_users
FROM user_growth
ORDER BY date;

2.3 窗口范围优化

累积聚合的性能关键在于窗口范围的优化。以下是几种常用的窗口范围定义：

```sql

ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW

```

```sql

ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING

```

```sql

ROWS BETWEEN 6 PRECEDING AND CURRENT ROW — 前6行到当前行

```

```sql

RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW

```

3. 分组 TopN 实现

分组 TopN 是窗口函数的另一个重要应用，用于在每个分组内获取前 N 条记录。相比传统的子查询或连接方式，使用窗口函数实现 TopN 更高效简洁。

3.1 基础实现方法

实现分组 TopN 的标准语法：

SELECT *
FROM (
SELECT
*,
ROW_NUMBER() OVER (
PARTITION BY 分组列
ORDER BY 排序列 DESC
) AS rn
FROM 表名
) ranked_data
WHERE rn <= N;

例如，获取每个产品类别销售额最高的前3个产品：

SELECT *
FROM (
SELECT
product_id,
product_name,
product_category,
sales_amount,
ROW_NUMBER() OVER (
PARTITION BY product_category
ORDER BY sales_amount DESC
) AS rn
FROM products
) ranked_products
WHERE rn <= 3
ORDER BY product_category, rn;

3.2 使用 RANK 和 DENSE_RANK

ROW_NUMBER()、RANK() 和 DENSE_RANK() 的区别：

• ROW_NUMBER()：为每一行分配唯一序号，不考虑并列
• RANK()：并列记录会得到相同排名，后续排名有空缺
• DENSE_RANK()：并列记录会得到相同排名，后续排名无空缺

例如，处理销售排名并列情况：

SELECT
product_id,
product_name,
product_category,
sales_amount,
RANK() OVER (
PARTITION BY product_category
ORDER BY sales_amount DESC
) AS sales_rank,
DENSE_RANK() OVER (
PARTITION BY product_category
ORDER BY sales_amount DESC
) AS dense_sales_rank
FROM products;

3.3 多级分组与 TopN

复杂业务场景中可能需要多级分组 TopN：

SELECT *
FROM (
SELECT
region,
city,
store_id,
sales_amount,
ROW_NUMBER() OVER (
PARTITION BY region, city
ORDER BY sales_amount DESC
) AS city_rank,
RANK() OVER (
PARTITION BY region
ORDER BY sales_amount DESC
) AS region_rank
FROM store_sales
) sales_ranked
WHERE city_rank <= 5 OR region_rank <= 10;

4. 复杂业务分析场景

窗口函数在实际业务中有着广泛的应用，特别是在需要复杂计算的场景中。本节将介绍几个高级应用案例。

4.1 同环比计算

计算同比（与去年同期相比）和环比（与上期相比）的变化：

SELECT
product_id,
month,
sales_amount,
LAG(sales_amount, 12) OVER (
PARTITION BY product_id
ORDER BY month
) AS year_ago_sales,
LAG(sales_amount, 1) OVER (
PARTITION BY product_id
ORDER BY month
) AS prev_month_sales,
(sales_amount – LAG(sales_amount, 12) OVER (
PARTITION BY product_id
ORDER BY month
)) / LAG(sales_amount, 12) OVER (
PARTITION BY product_id
ORDER BY month
) * 100 AS yoy_change,
(sales_amount – LAG(sales_amount, 1) OVER (
PARTITION BY product_id
ORDER BY month
)) / LAG(sales_amount, 1) OVER (
PARTITION BY product_id
ORDER BY month
) * 100 AS mom_change
FROM monthly_sales;

4.2 移动平均计算

计算移动平均是金融和销售分析中的常见需求：

SELECT
date,
value,
AVG(value) OVER (
ORDER BY date
ROWS BETWEEN 6 PRECEDING AND CURRENT ROW
) AS moving_avg_7days,
AVG(value) OVER (
ORDER BY date
ROWS BETWEEN 29 PRECEDING AND CURRENT ROW
) AS moving_avg_30days
FROM time_series_data;

4.3 分位数分析

分位数分析可用于用户行为分析、风险评估等场景：

SELECT
user_id,
purchase_amount,
NTILE(100) OVER (
ORDER BY purchase_amount
) AS percentile,
NTILE(4) OVER (
ORDER BY purchase_amount
) AS quartile
FROM user_purchases;

4.4 窗口函数嵌套使用

复杂场景下可以嵌套使用多个窗口函数：

WITH ranked_data AS (
SELECT
product_id,
sales_amount,
RANK() OVER (
PARTITION BY category
ORDER BY sales_amount DESC
) AS category_rank,
PERCENT_RANK() OVER (
PARTITION BY category
ORDER BY sales_amount
) AS sales_percentile
FROM products
)
SELECT
product_id,
sales_amount,
category_rank,
sales_percentile,
NTILE(5) OVER (
ORDER BY category_rank
) AS performance_tier,
CASE
WHEN category_rank <= 3 AND sales_percentile >= 0.8 THEN 'Top Performer'
WHEN category_rank <= 10 AND sales_percentile >= 0.6 THEN 'Good Performer'
ELSE 'Needs Attention'
END AS performance_flag
FROM ranked_data;

5. 最小示例与注意事项

5.1 最小可运行示例

以下是一个完整的 Spark SQL 窗口函数示例，可以直接运行：

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.window import Window

# 创建 SparkSession
spark = SparkSession.builder.appName("WindowFunctionExample").getOrCreate()

# 创建示例数据
data = [
("Electronics", "2023-01", 12000),
("Electronics", "2023-02", 15000),
("Electronics", "2023-03", 18000),
("Clothing", "2023-01", 8000),
("Clothing", "2023-02", 10000),
("Clothing", "2023-03", 12000)
]

columns = ["category", "month", "sales"]
df = spark.createDataFrame(data, columns)

# 定义窗口
window_spec = Window.partitionBy("category").orderBy("month")

# 应用窗口函数
result = df.select(
"category",
"month",
"sales",
F.sum("sales").over(window_spec).alias("cumulative_sales"),
F.row_number().over(window_spec).alias("month_rank"),
F.rank().over(window_spec).alias("sales_rank")
)

# 显示结果
result.show()

输出结果：

+———–+———+——+——————+———–+———–+
| category| month| sales|cumulative_sales|month_rank|sales_rank|
+———–+———+——+——————+———–+———–+
| Clothing|2023-01| 8000| 8000| 1| 1|
| Clothing|2023-02| 10000| 18000| 2| 2|
| Clothing|2023-03| 12000| 30000| 3| 3|
|Electronics|2023-01| 12000| 12000| 1| 1|
|Electronics|2023-02| 15000| 27000| 2| 2|
|Electronics|2023-03| 18000| 45000| 3| 3|
+———–+———+——+——————+———–+———–+

5.2 性能优化注意事项

5.3 兼容性注意事项