SparkStreaming 之 reduceByKeyAndWindow 详解及代码实现

摘要：reduceByKey 统计的是"当前这个 batch"，但"最近 10 分钟的热门词""最近 1 小时的 PV"这类需求要看一个滚动的时间窗口。reduceByKeyAndWindow 就是干这个的。这篇讲清 windowDuration 和 slideDuration 两个参数怎么理解，窗口重叠为什么会造成重复计算，以及两个重载版本（普通版 vs 带逆函数的增量版）的差别和选择。

关键词：Spark Streaming, reduceByKeyAndWindow, 滑动窗口, windowDuration, slideDuration, 逆函数

一、从 reduceByKey 到 reduceByKeyAndWindow

先回顾一个前提：reduceByKey 是每个 batch 独立聚合，只统计当前 2 秒（batchInterval）里的数据。但很多实时统计需求不是这样：

• 最近 10 分钟的热门搜索词。
• 最近 1 小时的页面 PV。
• 滑动窗口告警（比如"过去 5 分钟内错误数超过阈值就告警"）。

这些需求要看的是一段滚动的时间范围，reduceByKeyAndWindow 就是在一个"滑动窗口"上做 reduceByKey 聚合。

二、两个参数：windowDuration 和 slideDuration

理解窗口操作，先把这两个参数搞清楚：

pairs.reduceByKeyAndWindow(func, windowDuration, slideDuration)

• windowDuration（窗口长度）：一个窗口覆盖的时间范围，比如 10 秒。一个窗口 = 多个 batch 的并集。
• slideDuration（滑动间隔）：窗口多久滑动一次，比如 2 秒。也就是多久输出一次窗口结果。

举个例子：batchInterval = 2s，windowDuration = 10s，slideDuration = 4s，那么一个窗口覆盖 5 个 batch（10/2），每滑一次前进 2 个 batch（4/2）。

最容易踩的坑：windowDuration 和 slideDuration 都必须是 batchInterval 的整数倍。不满足这个约束，Spark 会在运行时报错。很多人配置窗口时随手写了个不是整数倍的秒数，调试半天才发现是这里的问题。

三、窗口重叠：数据被重复计算

窗口滑动的过程中，相邻窗口是重叠的。回到上面的例子，窗口 1 覆盖 batch0~batch4，窗口 2 覆盖 batch2_{batch6——batch2}batch4 被两个窗口共享。

这意味着：如果用最简单的实现，每个窗口都从零重新聚合，重叠部分的数据会被重复计算。重叠得越多，浪费越大。这正是后面"增量版"要解决的问题。

四、两个重载版本

reduceByKeyAndWindow 有两个重载，差别就在怎么处理重叠。

版本一：普通版（重复计算）

val result = pairs.reduceByKeyAndWindow(
(a: Int, b: Int) => a + b,
Seconds(10), Seconds(4)
)

每个窗口从零重新聚合，重叠部分重复算。实现简单，但窗口大、滑动小时浪费严重。小窗口、性能不敏感的场景够用。

版本二：增量版（带逆函数）

val result = pairs.reduceByKeyAndWindow(
(a: Int, b: Int) => a + b, // 累加函数：加新滑入的数据
(a: Int, b: Int) => a – b, // 逆函数：减滑出的数据
Seconds(10), Seconds(4)
)

增量版的思路是：新窗口 = 旧窗口 + 新滑入的 batch − 滑出的 batch。只算新增和移除，不重算重叠部分，性能大幅提升。

代价有两条：

五、完整代码：最近 10 分钟热门词

import org.apache.spark.streaming.{Seconds, Minutes, StreamingContext}

val ssc = new StreamingContext(conf, Seconds(2))

// 增量版必须开 checkpoint
ssc.checkpoint("hdfs://namenode:8020/checkpoint/hotword")

val lines = ssc.socketTextStream("localhost", 9999)

val hotWords = lines
.flatMap(_.split(" "))
.map(word => (word, 1))
.reduceByKeyAndWindow(
(a: Int, b: Int) => a + b, // _+_ 加新
(a: Int, b: Int) => a – b, // _-_ 减旧
Minutes(10), // 10 分钟窗口
Seconds(2) // 2 秒滑动
)

hotWords.print()
ssc.start(); ssc.awaitTermination()

这段代码每 2 秒输出一次"最近 10 分钟"的词频统计。_+_ 处理滑入窗口的新 batch，_-_ 处理滑出窗口的旧 batch，窗口里的重叠部分不重复计算。

六、总结

• reduceByKeyAndWindow 是在滑动窗口上做聚合，解决"最近 N 分钟"这类滚动统计需求。
• windowDuration 是窗口长度、slideDuration 是滑动间隔，两者都必须是 batchInterval 的整数倍。
• 窗口滑动会重叠，普通版会重复计算重叠数据；增量版用逆函数只算增量，性能更好。
• 增量版的两个前提：聚合函数有逆函数（加法↔减法、乘法↔除法，max/min 不行）、必须开 checkpoint。
• 窗口大、滑动小、性能敏感的场景优先增量版；小窗口或没有逆函数的聚合用普通版。

作者：大数据技术实践者 博客：blog.starzy.cn GitHub：starzy1990.github.io 专注 AI Agent · LangGraph · RAG · 大数据架构 · 数据工程实践