Flink水印基于事件时间。所谓事件时间,就是Flink DataStream中的数据元素自身带有的、在其实际发生时记录的时间戳,具有业务含义,并与系统时间独立。很显然,由于外部系统产生的数据往往不能及时,不能按序到达Flink系统,所以事件时间比处理时间有更强的不可预测性。为了能够准确地表达事件时间的处理进度,就必须用到水印。

Flink水印的本质是DataStream中的一种特殊元素,每个水印都携带有一个时间戳。当时间戳为T的水印出现时,表示事件时间t <= T的数据都已经到达,即水印后面应该只能流入事件时间t > T的数据。也就是说,水印是Flink判断迟到数据的标准,同时也是窗口触发的标记。

为了形象地说明水印的作用,参考一下下面的图,是一个乱序的基于事件时间的数据流示例。

flink.png

图中的方框就是数据元素,其中的数字表示事件时间,W(x)就表示时间戳是x的水印,并有长度为4个时间单位的滚动窗口。假设时间单位为秒,可见事件时间为2s、3s、1s的元素都会进入区间为[1s, 4s]的窗口,而事件时间为7s的元素会进入区间为[5s, 8s]的窗口。当水印W(4)到达时,表示已经没有t <= 4s的元素了,[1s, 4s]窗口会被触发并计算。同理,水印W(9)到达时,[5s, 8s]窗口会被触发并计算,以此类推。

不过图中暂时没有示出迟到数据。如果事件时间为6的元素出现在W(9)后面,就算是迟到了。迟到数据的处理后面再说。

迟到数据处理

如上所述,水印的乱序区间能够保证一些迟到数据不被丢弃,但是乱序区间往往不很长,那些真正迟到了的数据该怎么办呢?有两种方法来兜底,可以说是Flink为迟到数据提供的第二重保障。

窗口允许延迟

Flink提供了WindowedStream.allowedLateness()方法来设定窗口的允许延迟。也就是说,正常情况下窗口触发计算完成之后就会被销毁,但是设定了允许延迟之后,窗口会等待allowedLateness的时长后再销毁。在该区间内的迟到数据仍然可以进入窗口中,并触发新的计算。当然,窗口也是吃资源大户,所以allowedLateness的值要适当。给个完整的代码示例如下。

sourceStream.assignTimestampsAndWatermarks(
new BoundedOutOfOrdernessTimestampExtractor<UserActionRecord>(Time.seconds(30)) {
  private static final long serialVersionUID = 1L;
  @Override
  public long extractTimestamp(UserActionRecord record) {
    return record.getTimestamp();
  }
}
)
.keyBy("platform")
.window(TumblingEventTimeWindows.of(Time.minutes(1)))
.allowedLateness(Time.seconds(30))
.aggregate(new ViewAggregateFunc(), new ViewSumWindowFunc())
// ......

allowedLateness机制实际上就是DataFlow模型中的回填(backfill)策略的实现。对于滑动窗口和滚动窗口是累积(accumulating)策略,对于会话窗口则是累积与回撤(accumulating & retracting)策略。

侧输出迟到数据

侧输出(side output)是Flink的分流机制。迟到数据本身可以当做特殊的流,我们通过调用WindowedStream.sideOutputLateData()方法将迟到数据发送到指定OutputTag的侧输出流里去,再进行下一步处理(比如存到外部存储或消息队列)。代码如下。

// 侧输出的OutputTag
OutputTag<UserActionRecord> lateOutputTag = new OutputTag<>("late_data_output_tag");

sourceStream.assignTimestampsAndWatermarks(
new BoundedOutOfOrdernessTimestampExtractor<UserActionRecord>(Time.seconds(30)) {
  private static final long serialVersionUID = 1L;
  @Override
  public long extractTimestamp(UserActionRecord record) {
    return record.getTimestamp();
  }
}
)
.keyBy("platform")
.window(TumblingEventTimeWindows.of(Time.minutes(1)))
.allowedLateness(Time.seconds(30))
.sideOutputLateData(lateOutputTag)   // 侧输出
.aggregate(new ViewAggregateFunc(), new ViewSumWindowFunc())
// ......

// 获取迟到数据并写入对应Sink
stream.getSideOutput(lateOutputTag).addSink(lateDataSink);

标签: none

[2024年]-[红黑树技术联盟介绍]-[成员故事]


[2024年]-[JavaOffer]-[训练营]


添加新评论