Spark SQL中窗口函数与拆分处理性能对比分析

8 次阅读

Spark SQL中的窗口函数是处理分组内排序、排名、累计计算、前后记录关联等分析任务的重要工具。相比传统的拆分处理方式，窗口函数能够在单次SQL执行过程中完成分组与计算，但这并不意味着窗口函数在所有场景下都具有更高性能。尤其面对大规模数据、分区倾斜以及高基数分组时，窗口函数可能产生较大的Shuffle和排序开销。

因此，分析Spark SQL窗口函数与拆分处理的性能差异，需要从执行计划、Shuffle、排序、数据倾斜、任务数量以及内存消耗等多个维度进行判断。

一、窗口函数的基本工作机制

Spark SQL窗口函数通常通过OVER子句定义计算范围，例如：

SQL
SELECT
    user_id,
    event_time,
    amount,
    ROW_NUMBER() OVER (
        PARTITION BY user_id
        ORDER BY event_time
    ) AS rn
FROM user_events;

这里的PARTITION BY user_id用于划分窗口，ORDER BY event_time负责窗口内部排序，ROW_NUMBER()则根据排序结果生成序号。

常见窗口函数包括：

  • ROW_NUMBER()：生成唯一连续序号

  • RANK()：并列排名时保留间隔

  • DENSE_RANK()：并列排名但不跳号

  • LAG()：获取前一条记录

  • LEAD()：获取后一条记录

  • SUM()：计算累计值或窗口范围内的总和

  • AVG()：计算窗口平均值

  • FIRST_VALUE()：获取窗口中的首个值

  • LAST_VALUE()：获取窗口中的最后一个值

窗口函数最大的特点，是可以在不改变原始行粒度的情况下完成组内计算。

二、什么是拆分处理

所谓拆分处理，通常是先按照业务字段将数据拆分成不同的数据集，再分别完成计算，最后进行合并。

例如，需要找出每个用户最新的一条订单记录，可以通过窗口函数实现：

SQL
SELECT *
FROM (
    SELECT
        user_id,
        order_id,
        order_time,
        ROW_NUMBER() OVER (
            PARTITION BY user_id
            ORDER BY order_time DESC
        ) AS rn
    FROM orders
) t
WHERE rn = 1;

另一种思路是先按照用户进行拆分，再分别处理数据。

在实际Spark任务中，拆分处理可能表现为：

原始数据
   ↓
按照业务条件拆分
   ↓
多个子任务
   ↓
分别计算
   ↓
结果合并

这种方式在数据规模较小、业务规则差异明显时比较直观，但如果拆分数量过多，也可能产生大量任务调度和数据扫描开销。

三、窗口函数为什么可能产生较高性能开销

窗口函数看起来只是SQL中的一个函数，但Spark执行时通常需要完成分区和排序。

例如：

SQL
ROW_NUMBER() OVER (
    PARTITION BY department_id
    ORDER BY salary DESC
)

为了正确计算结果，Spark需要确保相同department_id的数据进入对应分区，并按照salary进行排序。

因此执行过程中可能出现：

Scan
 ↓
Exchange
 ↓
Sort
 ↓
Window

其中Exchange通常意味着Shuffle，而Sort则意味着排序成本。

如果数据量达到数亿甚至数十亿行，窗口函数的排序阶段可能成为整个任务的主要性能瓶颈。

四、拆分处理为什么有时更快

拆分处理并不一定意味着简单地把数据切成多个DataFrame。合理的拆分方式可以减少单次计算的数据规模。

例如原始数据按照业务类型分成：

type=A
type=B
type=C

如果三类数据的计算逻辑完全不同，那么直接使用多个针对性查询，可能比一个包含大量复杂窗口逻辑的SQL更容易优化。

另外，当某些数据本身具有明显的时间分区、业务分区或者物理分区特征时，拆分处理还可能利用分区裁剪减少扫描数据。

例如：

SQL
SELECT *
FROM orders
WHERE dt = '2026-09-30';

相比扫描整个订单表，再通过窗口函数处理某一天数据，提前进行过滤通常更有效。

但需要注意，拆分处理的优势来自“减少无效计算”，而不是来自“拆分”本身。如果只是机械地把一个任务拆成几十甚至几百个小任务，整体性能反而可能下降。

五、窗口函数与拆分处理的核心性能差异

两种方式的性能差异主要体现在以下几个方面。

1. Shuffle开销

窗口函数中的PARTITION BY可能触发数据重新分区。

例如：

SQL
ROW_NUMBER() OVER (
    PARTITION BY user_id
    ORDER BY event_time
)

如果user_id没有按照目标分区方式组织，Spark需要进行Shuffle。

拆分处理如果能够利用已有的数据分区，则可能减少这部分开销。

因此，判断性能时不能只看SQL语句本身，还要观察实际执行计划。

可以使用：

SQL
EXPLAIN FORMATTED
SELECT
    user_id,
    ROW_NUMBER() OVER (
        PARTITION BY user_id
        ORDER BY event_time
    ) AS rn
FROM events;

重点关注是否存在Exchange节点，以及Exchange前后的数据量。

2. 排序开销

窗口函数经常需要组内排序。

假设一个用户拥有1000万条行为记录，那么窗口计算可能需要处理一个非常大的分区。

如果业务实际上只需要最近一条数据，可以考虑先缩小数据范围，再执行窗口函数。

例如：

SQL
WITH filtered AS (
    SELECT *
    FROM events
    WHERE event_time >= '2026-09-01'
)
SELECT *
FROM (
    SELECT
        *,
        ROW_NUMBER() OVER (
            PARTITION BY user_id
            ORDER BY event_time DESC
        ) AS rn
    FROM filtered
) t
WHERE rn = 1;

先过滤再窗口计算，通常比对完整历史数据进行排序更合理。

3. 数据倾斜

数据倾斜是窗口函数性能问题中非常常见的一类。

例如：

SQL
PARTITION BY user_id

正常情况下，每个用户的数据量可能只有几百或几千条，但某个特殊用户拥有数千万条记录。

这个用户对应的Shuffle分区就会明显大于其他分区。

最终可能形成：

Task 1：100 MB
Task 2：120 MB
Task 3：110 MB
Task 4：12 GB

前三个任务很快结束，但最后一个任务长时间运行，导致整个Stage无法结束。

这种情况下，单纯增加Executor数量通常不能解决问题，因为数据集中在少数分区。

六、拆分处理也存在明显代价

拆分处理并非天然优于窗口函数。

如果将数据拆分成大量小任务，可能出现以下问题。

任务调度开销增加

假设原本只需要运行10个Task，却人为拆成1000个任务，那么Driver需要处理大量任务调度信息。

对于小数据集来说，真正耗时的可能已经不是计算，而是任务启动和调度。

重复读取数据

如果不同拆分逻辑需要重复扫描相同数据源，那么磁盘I/O和网络I/O可能显著增加。

例如：

任务A → 扫描orders
任务B → 扫描orders
任务C → 扫描orders

如果没有合理缓存或数据复用机制，拆分处理可能比单次扫描更加昂贵。

最终合并仍然需要Shuffle

拆分后如果最终还需要：

SQL
UNION ALL

或者重新进行：

SQL
GROUP BY
JOIN
ORDER BY

那么之前节省的部分计算成本，可能又被后续Shuffle抵消。

七、不同数据规模下的性能表现

窗口函数与拆分处理的性能不能脱离数据规模讨论。

小规模数据

当数据量只有几万到几百万行时，窗口函数通常具有较好的开发效率。

此时即使存在一定排序开销，也往往不会成为明显瓶颈。

相比手动拆分多个处理流程，窗口函数通常更加简洁。

中等规模数据

当数据量达到千万级以后，应重点检查：

  • Shuffle数据量

  • 排序耗时

  • 分区数量

  • 数据倾斜情况

  • Executor内存使用情况

此时需要通过执行计划和Spark UI判断瓶颈，而不能单纯依据SQL写法判断。

超大规模数据

当数据达到数亿或数十亿行时，窗口函数的代价会明显放大。

尤其是：

SQL
PARTITION BY 高基数字段
ORDER BY 时间字段

这种组合可能产生大量Shuffle和排序。

此时应该优先考虑数据预过滤、合理分区、控制Shuffle并行度以及处理倾斜数据。

八、窗口函数性能优化方法

1. 尽可能提前过滤数据

不要让窗口函数处理不必要的数据。

不推荐：

SQL
SELECT *
FROM (
    SELECT
        *,
        ROW_NUMBER() OVER (
            PARTITION BY user_id
            ORDER BY event_time DESC
        ) AS rn
    FROM events
) t
WHERE event_time >= '2026-09-01'
  AND rn = 1;

更合理的方式是：

SQL
SELECT *
FROM (
    SELECT
        *,
        ROW_NUMBER() OVER (
            PARTITION BY user_id
            ORDER BY event_time DESC
        ) AS rn
    FROM events
    WHERE event_time >= '2026-09-01'
) t
WHERE rn = 1;

这样可以让窗口计算面对更少的数据。

2. 选择合适的分区字段

PARTITION BY字段直接影响数据分布。

如果分区字段存在严重倾斜，应重新评估业务逻辑。

例如单纯按照：

SQL
PARTITION BY user_id

出现超级用户时，可以结合业务规则设计更合理的数据处理方式。

3. 合理设置Shuffle分区

Spark中的Shuffle分区数量会影响并行度。

可以通过：

SQL
SET spark.sql.shuffle.partitions=800;

进行调整。

分区过少，单个Task处理的数据量可能过大；分区过多，则会增加Task调度和管理成本。

因此不存在适用于所有任务的固定数值，需要结合数据规模和集群资源进行测试。

4. 避免无意义的全量排序

如果业务只是判断某个最大值或最小值，不一定需要窗口函数。

例如获取每个用户最新时间：

SQL
SELECT
    user_id,
    MAX(event_time) AS max_time
FROM events
GROUP BY user_id;

这种情况下，GROUP BY + MAX通常比完整窗口排序更加直接。

如果还需要返回对应记录，可以进一步考虑聚合与Join等方案。

九、什么时候应该优先选择窗口函数

以下场景通常比较适合窗口函数：

组内排名
组内Top N
获取上一条或下一条记录
累计计算
移动平均
连续事件分析
组内排序

例如：

SQL
SELECT
    user_id,
    order_id,
    amount,
    ROW_NUMBER() OVER (
        PARTITION BY user_id
        ORDER BY amount DESC
    ) AS rn
FROM orders;

如果业务逻辑本身就是典型的“组内计算”，窗口函数通常可以让SQL结构更加清晰。

十、什么时候可以考虑拆分处理

以下场景可以重点考虑拆分：

  1. 不同数据子集的处理逻辑完全不同。

  2. 数据天然按照时间或业务维度分区。

  3. 部分数据可以提前排除。

  4. 存在明显的数据倾斜。

  5. 不同数据集可以独立并行计算。

  6. 拆分后能够显著减少扫描和计算量。

例如日志分析中，如果历史数据和实时数据采用完全不同的计算策略，那么将二者拆开处理可能更加合理。

但拆分前需要确认不会引入大量重复扫描和额外Shuffle。

十一、使用Spark UI进行真实性能对比

性能分析不能只依赖理论推测，最可靠的方法是实际运行两个方案并查看Spark UI。

重点观察以下指标：

指标关注点
Input Size输入数据规模
Shuffle ReadShuffle读取量
Shuffle WriteShuffle写入量
Task Duration单Task执行时间
GC Time垃圾回收耗时
Spill Memory内存溢写情况
Spill Disk磁盘溢写情况
Failed Tasks是否存在异常Task

如果窗口函数方案的Shuffle Read明显增加，同时某几个Task运行时间远高于其他Task，通常需要进一步检查分区倾斜。

如果拆分方案产生大量短Task，则需要关注任务调度成本。

十二、一个简单的性能测试思路

可以准备相同数据，分别执行窗口函数方案和拆分方案。

例如测试每个用户最新订单。

窗口函数方案：

SQL
SELECT *
FROM (
    SELECT
        user_id,
        order_id,
        order_time,
        ROW_NUMBER() OVER (
            PARTITION BY user_id
            ORDER BY order_time DESC
        ) AS rn
    FROM orders
) t
WHERE rn = 1;

聚合方案则可以先得到最大时间：

SQL
WITH latest AS (
    SELECT
        user_id,
        MAX(order_time) AS max_order_time
    FROM orders
    GROUP BY user_id
)
SELECT o.*
FROM orders o
JOIN latest l
    ON o.user_id = l.user_id
   AND o.order_time = l.max_order_time;

两种方案最终结果可能接近，但执行成本并不一定相同。

尤其当存在订单时间重复时，第二种方式可能返回多条记录，而ROW_NUMBER()只会保留一条。因此性能测试必须建立在相同业务语义基础上，不能为了追求速度而改变结果逻辑。

十三、不要只比较执行时间

Spark任务的执行时间会受到很多因素影响，例如：

  • 集群当前负载

  • Executor数量

  • Executor内存

  • CPU资源

  • 数据缓存状态

  • 数据文件数量

  • 文件格式

  • 数据压缩方式

  • Shuffle分区数量

  • 数据倾斜程度

因此一次运行结果并不能完全代表实际性能。

更合理的方法是固定运行环境，多次测试，并同时比较：

总执行时间
CPU消耗
Shuffle数据量
内存占用
磁盘溢写
Task分布
资源利用率

只有综合这些指标，才能判断某种方案是否真正降低了计算成本。

十四、窗口函数与拆分处理的选择原则

实际项目中可以按照以下思路进行判断：

业务是否属于典型组内计算？
        ↓
       是
        ↓
优先考虑窗口函数
        ↓
检查Shuffle与排序成本
        ↓
是否存在明显倾斜？
   ↓             ↓
  否             是
  ↓              ↓
继续优化       重新设计分区或拆分逻辑

如果业务本身不需要排序，只需要分组统计，那么优先考虑聚合操作。

如果业务存在大量独立数据集，而且每个数据集的计算逻辑完全不同，可以考虑拆分处理。

如果只是为了规避窗口函数而盲目拆分，往往无法获得稳定的性能收益。

十五、总结

Spark SQL中的窗口函数和拆分处理并不存在绝对的性能优劣关系。窗口函数的优势在于表达能力强、SQL结构清晰，特别适合排名、Top N、前后记录关联和累计计算等场景；其主要性能成本则来自Shuffle、排序以及数据倾斜。

拆分处理可以通过缩小单次计算范围、利用数据分区和并行处理降低部分计算成本，但同时也可能带来重复扫描、任务调度以及结果合并等额外开销。

真正进行Spark SQL性能优化时，应当从数据规模、数据分布、业务语义和执行计划出发。对于窗口函数，重点关注Exchange、Sort、Shuffle以及倾斜分区；对于拆分方案，则需要关注任务数量、重复I/O和最终合并成本。

最可靠的优化方法不是简单地“避免窗口函数”，而是让计算路径与数据分布相匹配。通过EXPLAIN FORMATTED分析执行计划，再结合Spark UI中的Stage、Task和Shuffle指标进行实测，才能找到适合具体业务场景的高性能Spark SQL方案。

文章标签
标签1 Spark SQL 窗口函数 性能优化 数据处理