【Apache Flink】基于时间和窗口的算子-配置时间特性

前言

Apache Flink 它提供了多种类型的时间和窗口概念，使得用户能够进行准确的时间计算。在数据处理任务中，时间的概念是非常重要的，对于一些复杂的实时流处理任务，如事件按时间顺序的聚合、分割和窗口计算，时间更是关键所在。而在这类任务中，选择使用何种时间特性是决定结果准确性的非常重要的一部分。Flink提供了三种时间特性供用户选择：事件时间、处理时间和摄取时间。

在使用Flink进行流处理时，时间窗口的选择也至关重要。Flink的窗口操作可以帮助我们将无限的流数据切分为有限的块，方便我们对每个数据块进行计算。Flink提供了多种窗口类型，包括滑动窗口、滚动窗口、会话窗口和全局窗口，用户可以根据业务需求选择合适的窗口类型。

本文将详细介绍Apache Flink中基于时间和窗口的算子，以及如何配置数据流的时间特性， 深入地理解和使用Flink在实际流式处理任务中进行数据计算的能力。

配置时间特性

在Apache Flink中，时间的概念是极其关键的要素，尤其是当我们处理实时或近实时数据流的时候。Flink提供了三种不同的时间概念，分别用于处理不同的任务和场景。

1. 事件时间 (Event Time): 事件时间是指数据产生的时间，这个时间通常由事件数据本身携带，例如每个事件的日志行通常都会记录时间戳。在处理可能存在乱序或延迟数据的流计算任务时，事件时间是最常用的处理策略，因为它可以按照事件的发生顺序处理信息，而不是它们被系统接收的顺序。

2. 处理时间 (Processing Time): 处理时间则是事件在系统中被处理时的系统时间，即事件在引擎处理时的“现在”时间。处理时间对于想要最大性能、毫秒级结果的场景非常适用，比如实时监控或者实时报警这样对处理速度有极高要求的场景。

3. 摄取时间 (Ingestion Time): 摄取时间属于事件时间和处理时间的一种折衷方案。它是指事件进入Flink系统的时间。如果在一些场合，事件的时间戳无法获取或者不准确，同时又需要一定的事件排序，那么摄取时间就派上用场了。其在源头就分配了时间戳，并在整个处理过程中保留。摄取时间比处理时间语义强，比事件时间性能好。

这三种时间概念的选用，在很大程度上，决定了Flink处理事件的顺序和方式，因此根据实际的场景选择最适合的时间策略是非常重要的。

1. 事件时间 (Event Time)

  env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

Flink会根据事件所携带的时间戳来处理数据，这就说明了数据是基于何时发生的进行处理的，而不是基于何时被处理的。比如在处理日志分析时，如果日志已经按照事件的发生时间排序，那么事件时间这种设置就会非常有用。

2. 处理时间 (Processing Time)

   env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);

Flink会按照数据进入系统的时间即系统处理的实时进行处理，无视事件自身的时间戳。比如对于实时监控或者实时报警这样对处理速度有极高要求的场景，处理时间是最适合的。

3. 摄取时间 (Ingestion Time)

  env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);

在数据读入Flink时，会自动地获取数据的摄取时间，与处理时间的观念类似但在数据进入系统时就已经赋予了时间。比如在希望事件能在一定程度上按照顺序处理，但又无法获取准确的事件时间时，摄取时间是一个不错的选择。

将时间特性设置为事件时间

在 Flink 中，我们可以指定时间特性为事件时间（Event Time），这是处理一些具体问题（例如延迟数据、重新处理等）的关键。下面是如何在 Flink 中设置时间特性为事件时间的代码示例：
读入一些元素，为它们分配时间戳和水印，并打印出来。分配时间戳和水印是在使用事件时间进行窗口操作等处理时必须要做的。

创建一个执行环境，并设置为事件时间。然后创建一个简单的字符串流，并分配时间戳和水印。最后我们使用了一个时间窗口并打印出了所有的元素。。

import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;public class EventTimeExample {public static void main(String[] args) throws Exception {// 创建执行环境final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 设置时间特性为事件时间env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);// 创建一个数据源，这里我们用的是从集合中获取DataStream<String> stream = env.fromElements("element1", "element2", "element3");// 分配时间戳和水印，这里我们假设通过某种方法获取了一个递增的时间戳DataStream<String> withTimestampsAndWatermarks = stream.assignTimestampsAndWatermarks(new AscendingTimestampExtractor<String>() {@Overridepublic long extractAscendingTimestamp(String element) {// 返回当前元素的时间戳，这儿仅作示例，实际应用需要根据具体需求获取return System.currentTimeMillis();}});// 对数据进行一些操作，比如过滤，这里的窗口大小为5秒withTimestampsAndWatermarks.timeWindowAll(Time.seconds(5)).apply(new AllWindowFunction<String, Object, TimeWindow>() {// 在应用函数中，我们做一些处理，这里简单地打印出所有元素public void apply(TimeWindow window, Iterable<String> values, Collector<Object> out) {for (String value : values) {System.out.println(value);}}});// 启动应用env.execute("Event Time Example");}
}

时间戳分配器

在数据处理系统中，特别是在事件驱动的系统或实时流处理系统中，时间戳分配器是一个非常重要的组件。时间戳分配器的任务就是为每一个事件或数据记录分配一个时间戳，这个时间戳表示该事件发生的时间。

在 Apache Flink 这样的流处理框架中，时间戳分配器通常在数据源（Source）接收到数据时运行，为每一个事件分配一个时间戳。基于事件时间处理的窗口运算、Watermark 生成等流处理任务依赖于这个时间戳。

提供一个基础的时间戳分配器示例，此处假设有一个MapFunction，它将输入的MyEvent转换为一个包含时间戳和事件数据的Tuple2：

MyTimestampAssigner实现了AssignerWithPunctuatedWatermarks接口，extractTimestamp()方法用于从数据元素中抽取出时间戳，checkAndGetNextWatermark()方法用于生成水印。在这里每处理一个元素都会调用checkAndGetNextWatermark()生成一个新的水印，这被称为 “punctuated” 模式。若要使用更常见的 “periodic” 模式，需要实现AssignerWithPeriodicWatermarks接口，这会定期（而不是每处理一个元素）生成水印。

public class MyTimestampAssigner implements AssignerWithPunctuatedWatermarks<MyEvent> {@Nullable@Overridepublic Watermark checkAndGetNextWatermark(MyEvent event, long extractTimestamp) {return new Watermark(extractTimestamp);}@Overridepublic long extractTimestamp(MyEvent element, long previousElementTimestamp) {return element.getCreationTime();}
}

然后，可以在流处理中使用这个MyTimestampAssigner来为数据流中的每一个元素分配时间戳：

DataStream<MyEvent> stream = ...
stream.assignTimestampsAndWatermarks(new MyTimestampAssigner());

---

源数据通常会有一个时间字段，它代表了数据的生成时间。我们可以基于这个时间字段来处理数据，并生成结果。为了让Flink知道每条数据的时间，我们需要自定义Timestamps/Watermarks。以下是一个简单的示例，该代码是在Flink中进行窗口计数的常见工作模式。

在水印生成部分设置了10秒的延迟时间，以处理乱序数据。这意味着，当窗口看到最新的水印时间大于其结束时间时，窗口才会触发进行执行计算。触发后的10秒内，该窗口还会接收更晚到达的数据。这对于处理乱序数据非常有用。

import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.api.java.tuple.Tuple2;public class TimeStampWaterMarkExample {public static void main(String[] args) throws Exception {// 创建执行环境，设置为事件时间模式StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);// 元组的第一个参数是一个字符串，第二个参数是一个时间戳Tuple2[] data = new Tuple2[]{new Tuple2<>("a", 1558430842000L),new Tuple2<>("b", 1558430843000L),new Tuple2<>("c", 1558430845000L)};// 添加数据源DataStreamSource<Tuple2<String, Long>> dataSource = env.addSource(new SourceFunction<Tuple2<String, Long>>() {@Overridepublic void run(SourceContext<Tuple2<String, Long>> ctx) throws Exception {for (Tuple2<String, Long> datum : data) {ctx.collectWithTimestamp(datum, datum.f1);ctx.emitWatermark(new Watermark(datum.f1 - 1));}ctx.emitWatermark(new Watermark(Long.MAX_VALUE));}@Overridepublic void cancel() {}});// 添加我们的时间戳分配器和水印生成器SingleOutputStreamOperator<Tuple2<String, Long>> timestampOperator = dataSource.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Tuple2<String, Long>>(Time.seconds(10)) {@Overridepublic long extractTimestamp(Tuple2<String, Long> element) {return element.f1;}});// 执行滚动窗口操作timestampOperator.keyBy(0).window(TumblingEventTimeWindows.of(Time.seconds(2)))  // 定义滚动窗口.apply(new CountWindowFunction()).print();env.execute("job name");}
}

周期性水位线分配器

在流处理系统中，Watermark（水位线）是一种用于处理事件时间乱序的机制。在实时流处理系统中，来自不同源的事件可能以不同的顺序到达。这种顺序的不确定性会给处理系统带来挑战，因为系统必须确定何时所有相关事件都已到达，以便可以进行处理（例如，计算一个5分钟滑动窗口的平均值）。

Watermark就是流处理系统用于处理这种不确定性的一种手段。在Flink中，Watermark是一个特殊的事件或者信号，表示在此时间戳之前的所有事件都已经到达，没有更早的事件会到达。当Watermark t 到达时，说明在时间 t 或者更早的所有数据都已经收到，可以开始处理时间小于或等于 t 的窗口。

周期性水位线 (Periodic Watermark) 是一种定时生成的 Watermark。具体来说，数据流会周期性地调用 getCurrentWatermark() 方法获取新的 Watermark，并且插入到流中。这个周期默认是200ms，也可以通过 ExecutionConfig.setAutoWatermarkInterval(…) 来设置。

在定义 Watermark 生成逻辑时，通常会设置一个允许乱序到达的事件的最大延迟时间。例如，如果最大延迟时间设置为5秒，那么就意味着，Watermark t 只能保证时间戳小于 t-5s 的所有事件都已经到达。换句话说，水位线是滞后于事件时间的，即使水位线是周期性地生成的。

创建一个实现AssignerWithPeriodicWatermarks接口的类，目的是为了周期性生成watermark

BoundedOutOfOrdernessGenerator实现了AssignerWithPeriodicWatermarks接口。extractTimestamp()方法提取了每个元素的时间戳，getCurrentWatermark()则返回新的 watermark。这个实现假设元素可以最多晚3.5秒到达。使用当前最大时间戳减去这个延迟就得到了新的 watermark。这就意味着，即使有延迟的元素到达，只要其时间戳比当前的 watermark 大，它仍然可以用于窗口计算。

 
public class BoundedOutOfOrdernessGenerator implements AssignerWithPeriodicWatermarks<MyEvent> {// 最大的乱序容忍度设置为3500毫秒，也就是3.5秒private final long maxOutOfOrderness = 3500; // 当前收到的最大的时间戳private long currentMaxTimestamp;@Override// 此方法用于从数据元素中提取时间戳public long extractTimestamp(MyEvent element, long previousElementTimestamp) {long timestamp = element.getCreationTime();// 更新当前收到的最大的时间戳currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp);// 返回提取的时间戳return timestamp;}@Nullable@Override// 此方法返回当前的Watermark，Watermark值为收到的最大时间戳减去乱序容忍度public Watermark getCurrentWatermark() {return new Watermark(currentMaxTimestamp - maxOutOfOrderness);}
}

在流处理中用这个BoundedOutOfOrdernessGenerator为流中的每个元素分配时间戳，并定期生成水位线。

DataStream<MyEvent> stream = ...
// 对流中的元素调用assignTimestampsAndWatermarks函数，传入的参数为BoundedOutOfOrdernessGenerator的实例，这样就完成了对流中元素的时间戳分配和周期性的Watermark生成
stream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessGenerator());

定点水位线分配器

定点水位线分配器（Fixed Watermark Assigner）是一种特定类型的水位线分配器，它生成一个固定的水位线值。它通常在事件的时间戳已经有序且无乱序出现时使用。由于每个元素都有一个时间戳，在这种情况下，水位线可能仅被推进到元素的当前时间戳。在一些特定的应用场景中，这样可能就够用了。

在Apache Flink中，可以使用assignTimestampsAndWatermarks函数结合WatermarkStrategy实现水位线的分配。对于定点水位线分配器，可以创建一个固定的，不变的水位线，例如：

WatermarkStrategy.<YourEvent>forMonotonousTimestamps().withTimestampAssigner((event, timestamp) -> event.getTimestamp());

在这个例子中，forMonotonousTimestamps方法会创建一个定点水位线分配器，它将水位线固定在最近被处理的事件的时间戳。withTimestampAssigner方法定义了如何从事件中抽取时间戳。

定点水位线分配器用在那些时间戳严格递增的事件流中，如果在这样的流上使用定点水位线分配器，如果事件到达的顺序与时间戳的顺序不一致（例如，由于网络延迟、机器间的时钟偏移等），就有可能会得到错误的结果。

示例

在这个示例中假定事件是按照时间戳顺序处理的，所以可以有效地使用定点水位线分配器。如果事件的顺序可能会被打乱，那建议使用其他类型的水位线分配器。

创建一个水位线函数，对每一个输入的元素都分配它的时间戳，并产生连续定点的水位线。

public static class MyEvent {public long timestamp;  // 定义保存时间戳的变量public String data;  // 定义保存数据的变量public MyEvent(long timestamp, String data) {  // 构造函数this.timestamp = timestamp;this.data = data;}public long getTimestamp() {  // 获取时间戳的接口return timestamp;}
}

定义了事件类，包括时间戳和数据。

import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;// 初始化执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 配置事件时间特性
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);// 输入的数据流
DataStream<MyEvent> inputStream = env.fromElements(new MyEvent(System.currentTimeMillis(), "data1"),new MyEvent(System.currentTimeMillis() + 1000, "data2"),new MyEvent(System.currentTimeMillis() + 2000, "data3")
);// 使用水位线分配器给事件分配时间戳和水位线
DataStream<MyEvent> withTimestampsAndWatermarks = inputStream.assignTimestampsAndWatermarks(WatermarkStrategy.<MyEvent>forMonotonousTimestamps().withTimestampAssigner((event, timestamp) -> event.getTimestamp()));

参考文档

1. https://nightlies.apache.org/flink/flink-docs-release-1.18/zh/