当前位置：首页 > news >正文

Flink DataStream API深度解析（Scala版）：窗口计算、水位线与状态编程

news 来源：原创 2025/8/28 19:06:12

在前面的文章中Flink 编程基础：Scala 版 DataStream API 入门-CSDN博客，我们已经介绍了 Flink 的 Datastream API 编程模型、窗口划分以及时间语义（处理时间、事件时间等）。本篇文章将深入讲解窗口计算的进阶内容，包括水位线（Watermark）、延迟数据处理（Late Data Handling）以及状态编程（State Programming）。通过手把手的示例，我们将逐步揭开这些概念的神秘面纱，帮助大家构建健壮的流处理应用程序。

一、窗口计算

1.1 窗口分配器（Window Assig2ner）

滚动窗口（Tumbling Window）

滑动窗口（Sliding Window）

会话窗口（Session Window）

1.2 窗口计算函数

ReduceFunction（增量计算）

AggregateFunction（可控的增量计算）

ProcessWindowFunction（全量计算）

组合用法

1.3 触发器 Trigger

1.4 驱逐器 Evictor

二、水位线（Watermark）详解

三、延迟数据处理

3.1 allowedLateness：允许一定时间内迟到的数据

3.2 sideOutputLateData：输出极度迟到数据

四、状态编程

4.1 状态类型

4.2 示例：使用 ValueState

结语

一、窗口计算

在流式处理中，由于数据是无界的，我们需要对数据进行分组（按照时间或数量切分）来进行聚合计算。窗口计算（Window Computation）的主要思想是对数据按照一定规则进行划分，然后在每个窗口内执行计算操作，比如求和、求平均、计算最大最小值等。

Flink 中的窗口计算是流处理中的核心操作，用于将无限的数据流划分为有限的、可计算的块（窗口）。窗口计算的一般结构如下：

DataStream → keyBy() → window()/windowAssigner 
→ allowedLateness → trigger → evictor 
→ apply/reduce/aggregate → Sink

我们可以把窗口看作是给数据加上“时间范围”的过滤器，把连续流拆分成一个个小段，每段再单独进行聚合或统计处理。

1.1 窗口分配器（Window Assig2ner）

窗口分配器负责告诉 Flink 如何将数据划分到一个个窗口中。主要有三种常见窗口类型：

滚动窗口（Tumbling Window）

滚动窗口的特点是固定时间长度，并且窗口之间互不重叠。每个元素只属于一个窗口。

.window(TumblingEventTimeWindows.of(Time.seconds(10)))

上面的代码表示每 10 秒划分一个窗口，例如 [00:00:00, 00:00:10), [00:00:10, 00:00:20) 等。

上图展示了 Flink DataStream API 的滚动窗口 机制：

时间轴（t0 至 t4）上严格划分出固定大小的连续窗口（窗口1至窗口4），每个窗口独立且不重叠，事件数据（●）根据时间戳归属到唯一对应的窗口中，适用于周期性的无重复聚合计算（如每5秒统计流量）。

滑动窗口（Sliding Window）

滑动窗口支持窗口重叠。可以设定窗口大小和滑动步长。

.window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))

此处每 5 秒滑动一次，窗口长度为 10 秒，例如：

[00:00:00, 00:00:10)
[00:00:05, 00:00:15)
[00:00:10, 00:00:20)

此图展示了Flink DataStream API 中的滑动窗口机制。图中设定窗口大小为2，滑动步长为1 ，意味着每1个时间单位会计算最近2个时间单位的数据。通过窗口1（[t0,t2] ）、窗口2（[t1,t3] ）、窗口3（[t2,t4] ）的范围示意窗口滑动过程，重叠区域体现窗口重叠特性，表明存在重复计算数据的情况。

蓝色点：窗口 1 范围是 [t0,t2) ，它在这个区间内；窗口 2 范围是 [t1,t3] ，当时间走到 t1 ，窗口 2 开启，蓝色点所处时间也在窗口 2 的时间跨度里，所以它同时属于窗口 1 和窗口 2 。

绿色点：窗口 1 是左闭右开区间 [t0,t2) ，不包含 t2 ，绿色点虽在 t1 - t2 之间，但没到 t2 ，就不在窗口 1 里；而窗口 2 范围是 [t1,t3] ，把绿色点 “包” 进去了，所以绿色点只属于窗口 2

会话窗口（Session Window）

会话窗口不是固定时间，而是基于事件之间的不活动间隔进行划分。

.window(ProcessingTimeSessionWindows.withGap(Time.seconds(5)))

如果两个事件之间的时间间隔超过 5 秒，就会触发一个新窗口。

1.2 窗口计算函数

窗口划分好之后，我们就需要对每个窗口中的元素进行计算。这可以通过三种窗口函数实现：

ReduceFunction（增量计算）

适用于可以连续合并的数据，如求和、最大值等。

.reduce((v1, v2) => (v1._1, v1._2 + v2._2))

上面这个例子是对每个 key 的值进行累加。

AggregateFunction（可控的增量计算）

更复杂的增量计算方式，支持中间状态（Accumulator）。

class MyAgg extends AggregateFunction[(String, Int), Int, Int] {override def createAccumulator(): Int = 0override def add(value: (String, Int), acc: Int): Int = acc + value._2override def getResult(acc: Int): Int = accoverride def merge(acc1: Int, acc2: Int): Int = acc1 + acc2
}

ProcessWindowFunction（全量计算）

拥有更强大的上下文信息，可以访问窗口元数据和所有元素。

class MyWindowFunc extends ProcessWindowFunction[(String, Int), String, String, TimeWindow] {override def process(key: String, context: Context, elements: Iterable[(String, Int)], out: Collector[String]): Unit = {val sum = elements.map(_._2).sumout.collect(s"$key 窗口 [${context.window.getStart} - ${context.window.getEnd}] 总和: $sum")}
}

组合用法

可将增量计算与全量处理结合，效率更高：

.aggregate(new MyAgg, new MyWindowFunc)

1.3 触发器 Trigger

触发器决定何时执行窗口计算逻辑。默认是事件时间触发：

.trigger(CountTrigger.of(5))

表示每收到 5 条数据就触发一次计算。常用触发器类型：

EventTimeTrigger：基于水位线，默认类型。

ProcessingTimeTrigger：基于系统时间。

CountTrigger：基于数据条数。

自定义 Trigger：继承 Trigger 类实现自定义逻辑。

1.4 驱逐器 Evictor

Evictor 控制在窗口计算之前剔除一部分元素，适用于对窗口中数据量进行限制。

.evictor(CountEvictor.of(3))

表示只保留最后 3 个元素进行窗口计算。

二、水位线（Watermark）详解

在处理事件时间窗口时，由于数据可能无序到达，需要用水位线机制来判断数据是否已经“齐全”。水位线是一种特殊的时间标记，用来指出目前事件时间的进度。当数据的事件时间超过水位线后，窗口操作就可以安全地触发计算。

如何定义水位线

单调递增时间戳（适合有序数据）：

data.assignAscendingTimestamps(_.timestamp)

有限乱序时间戳（容忍延迟）：

data.assignTimestampsAndWatermarks(WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(5)).withTimestampAssigner(new SerializableTimestampAssigner[MyEvent] {override def extractTimestamp(element: MyEvent, recordTimestamp: Long): Long = element.timestamp}))

上面的策略会容忍最多 5 秒的乱序。

示例：

import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks
import org.apache.flink.streaming.api.watermark.Watermarkclass MyTimestampExtractor extends AssignerWithPeriodicWatermarks[(String, Long)] {// 定义允许的最大乱序间隔（毫秒），比如 3 秒val maxOutOfOrderness = 3000Lvar currentMaxTimestamp: Long = _override def extractTimestamp(element: (String, Long), previousElementTimestamp: Long): Long = {val timestamp = element._2currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp)timestamp}override def getCurrentWatermark: Watermark = {// 当前水位线：当前最大时间戳减去乱序间隔new Watermark(currentMaxTimestamp - maxOutOfOrderness)}
}

讲解：

我们通过 maxOutOfOrderness 定义允许数据乱序的最大容忍时间。

每次取到元素时更新 currentMaxTimestamp，水位线则为 currentMaxTimestamp - maxOutOfOrderness。

这样就可以确保，当事件时间超过水位线后，窗口内数据已经尽可能到齐，可以放心触发计算。

将上述时间戳提取器应用到数据流中：

val timestampedStream = dataStream.assignTimestampsAndWatermarks(new MyTimestampExtractor)

三、延迟数据处理

在处理流数据时，迟到的数据是不可避免的。Flink 提供了三种策略来应对：

3.1 allowedLateness：允许一定时间内迟到的数据

.allowedLateness(Time.seconds(60))

这个设置告诉 Flink：即使窗口已经关闭，在接下来的 60 秒内到来的迟到数据依然会重新触发计算。

3.2 sideOutputLateData：输出极度迟到数据

.sideOutputLateData(lateOutputTag)

对于超过 allowedLateness 时间的极度迟到数据，我们可以将其发送到旁路输出流中，从而单独处理。

四、状态编程

Flink 强大的地方在于它是有状态的流处理框架。状态用于记录和保存处理过程中的中间结果。

4.1 状态类型

状态类型	描述
ValueState	保存单个值
ListState	保存多个值（列表）
MapState	保存键值对（K-V）
ReducingState	自动合并的聚合状态
BroadcastState	广播状态，所有子任务共享

4.2 示例：使用 ValueState

以下示例展示如何统计每个 key 的累加值：

class StatefulMapFunction extends RichMapFunction[(String, Int), (String, Int)] {private var countState: ValueState[Int] = _override def open(parameters: Configuration): Unit = {val desc = new ValueStateDescriptor[Int]("count", createTypeInformation[Int])countState = getRuntimeContext.getState(desc)}override def map(value: (String, Int)): (String, Int) = {val current = Option(countState.value()).getOrElse(0) + value._2countState.update(current)(value._1, current)}
}

使用状态时有两个要点：

必须在 keyBy 后使用状态，状态与 key 绑定。
Flink 会自动进行状态的 checkpoint 和恢复。

结语

本文系统介绍了 Flink Scala 版 DataStream API 中窗口计算的各个组件、水位线与迟到数据处理机制，以及核心的状态编程能力。掌握这些内容后，将能够构建更加鲁棒、高性能的流处理程序。

如果这篇文章对你有所启发，期待你的点赞关注！

一、窗口计算

1.1 窗口分配器（Window Assig2ner）

滚动窗口（Tumbling Window）

滑动窗口（Sliding Window）

会话窗口（Session Window）

1.2 窗口计算函数

ReduceFunction（增量计算）

AggregateFunction（可控的增量计算）

ProcessWindowFunction（全量计算）

组合用法

1.3 触发器 Trigger

1.4 驱逐器 Evictor

二、水位线（Watermark）详解

三、延迟数据处理

3.1 allowedLateness：允许一定时间内迟到的数据

3.2 sideOutputLateData：输出极度迟到数据

四、状态编程

4.1 状态类型

4.2 示例：使用 ValueState

结语

相关文章：