文章目录
1. Flink 中的时间语义
在Flink的流式处理中,会涉及到时间的不同概念,如下图所示:
-
Ingestion Time:是数据进入Flink的时间。
-
Processing Time:是每一个执行基于时间操作的算子的本地系统时间,与机器相关,默认的时间属性就是ProcessingTime。
例如,一条日志进入 Flink 的时间为2020-09-03 10:00:00.123
,到达Window的系统时间为2020-09-03 10:00:01.234
,日志的内容如下:
2020-09-03 09:59:58.624 INFO Fail over to rm2
对于业务来说,要统计 1 min
内的故障日志个数,哪个时间是最有意义的?—— eventTime,因为我们要根据日志的生成时间进行统计。
2. EventTime的引入
在Flink的流式处理中,绝大部分的业务都会使用 eventTime,一般只在 eventTime 无法使用时,才会被迫使用 ProcessingTime 或者 IngestionTime。
如果要使用 EventTime,那么需要引入 EventTime 的时间属性,引入方式如下所示:
val env = StreamExecutionEnvironment.getExecutionEnvironment
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
3. Watermark
3.1 基本概念
我们知道,流处理从事件产生,到流经 source,再到 operator,中间是有一个过程和时间的,虽然大部分情况下,流到 operato r的数据都是按照事件产生的时间顺序来的,但是也不排除由于网络、分布式等原因,导致乱序的产生,所谓乱序,就是指 Flink 接收到的事件的先后顺序不是严格按照事件的 EventTime 顺序排列的。
- Watermark 是一种衡量 Event Time 进展的机制。
- Watermark 是用于处理乱序事件的,而正确的处理乱序事件,通常用 Watermark 机制结合 window 来实现。
- 数据流中的 Watermark 用于表示 timestamp 小于 Watermark 的数据,都已经到达了,因此,window 的执行也是由 Watermark 触发的。
- Watermark 可以理解成一个延迟触发机制,我们可以设置 Watermark 的延时时长t,每次系统会校验已经到达的数据中最大的 maxEventTime,然后认定 eventTime 小于 maxEventTime-t 的所有数据都已经到达,如果有窗口的停止时间等于 maxEventTime–t,那么这个窗口被触发执行。
有序流的 Watermarker 如下图所示:(Watermark 设置为0):
上图中,我们设置的允许最大延迟到达时间为 2s,所以时间戳为 7s的事件对应的Watermark是5s,时间戳为12s的事件的 Watermark 是 10s,如果我们的窗口1
是 1s~5s
,窗口2
是6s~10s,那么时间戳为7s
的事件到达时的 Watermarker 恰好触发窗口1
,时间戳为12s
的事件到达时的Watermark恰好触发窗口2
。
Watermark 就是触发前一窗口的“关窗时间”,一旦触发关门那么以当前时刻为准在窗口范围内的所有所有数据都会收入窗中。
只要没有达到水位那么不管现实中的时间推进了多久都不会触发关窗。
3.2 WaterMark的引入
Watermark 的引入很简单,对于乱序数据,最常见的引用方式如下:
stream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(2)){
override def extractTimestamp(element: SensorReading) = {
element.timestamp * 1000
}
})
Event Time 的使用一定要指定数据源中的时间戳。否则程序无法知道事件的事件时间是什么(数据源里的数据没有时间戳的话,就只能使用 Processing Time 了)。
我们看到上面的例子中创建了一个看起来有点复杂的类,这个类实现的其实就是分配时间戳的接口。Flink 暴露了 TimestampAssigner 接口供我们实现,使我们可以自定义如何从事件数据中抽取时间戳。
val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
val readings:DataStream[SensorReading]=env
.addSource(newSensorSource)
.assignTimestampsAndWatermarks(newMyAssigner())
MyAssigner 有两种类型:
-
AssignerWithPeriodicWatermarks
-
AssignerWithPunctuatedWatermarks
以上两个接口都继承自TimestampAssigner
。
3.2.1 周期性生成 Watermark
周期性的生成 Watermark:系统会周期性的将 watermark 插入到流中(水位线也是一种特殊的事件!)。默认周期是200毫秒。可以使用ExecutionConfig.setAutoWatermarkInterval()
方法进行设置。
env.getConfig.setAutoWatermarkInterval(5000)
产生 watermark 的逻辑:每隔5秒钟,Flink会调用AssignerWithPeriodicWatermarks
的getCurrentWatermark()
方法。如果方法返回一个时间戳大于之前水位的时间戳,新的 watermark 会被插入到流中。这个检查保证了水位线是单调递增的。如果方法返回的时间戳小于等于之前水位的时间戳,则不会产生新的watermark。
例子,自定义一个周期性的时间戳抽取:
package com.flink.scala
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks
import org.apache.flink.streaming.api.watermark.Watermark
/**
* 自定义周期性的时间戳抽取类
*/
class PeriodicAssigner extends AssignerWithPeriodicWatermarks[SensorReading] {
val bound: Long = 60 *1000 //延时 1 分钟
var maxTs: Long = Long.MinValue // 观察到的最大时间戳
override def getCurrentWatermark: Watermark = {
new Watermark(maxTs - bound)
}
override def extractTimestamp(t: SensorReading, l: Long): Long = {
maxTs = maxTs.max(t.timestamp)
t.timestamp
}
}
一种简单的特殊情况是,如果我们事先得知数据流的时间戳是单调递增的,也就是说没有乱序,那我们可以使用assignAscendingTimestamps
,这个方法会直接使用数据的时间戳生成 watermark。
val withTimestampsAndWatermarks = stream.assignAscendingTimestamps(e=>e.timestamp)
而对于乱序数据流,如果我们能大致估算出数据流中的事件的最大延迟时间,就可以使用如下代码:
val withTimestampsAndWatermarks = stream.assignTimestampsAndWatermarks(newSensorTimeAssigner)
class SensorTimeAssigner extends
BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(5)){
//抽取时间戳
override def extractTimestamp(r:SensorReading):Long=r.timestamp
}
3.2.2 间断式生成 Watermark
间断式的生成 watermark 和周期性生成的方式不同,这种方式不是固定时间的,而是可以根据需要对每条数据进行筛选和处理。直接上代码来举个例子,我们只给 sensor_1 的传感器的数据流插入watermark:
package com.flink.scala
import org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks
import org.apache.flink.streaming.api.watermark.Watermark
/**
* 间断式生成 Watermark
*/
class PunctuatedAssigner extends AssignerWithPunctuatedWatermarks[SensorReading] {
val bound:Long=60*1000
override def checkAndGetNextWatermark(t: SensorReading, extractedTS: Long): Watermark = {
if ("sensor_1".equals(t.id)) {
new Watermark(extractedTS - bound)
} else {
null
}
}
override def extractTimestamp(t: SensorReading, l: Long): Long = {
t.timestamp
}
}
4. EvnetTime 在 window 中的使用
4.1 滚动窗口(TumblingEventTimeWindows)
package com.flink.scala
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.{
DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import scala.collection.mutable
object TumblingEventTimeWindowsTest {
def main(args: Array[String]): Unit = {
val env = StreamExecutionEnvironment.getExecutionEnvironment
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
env.setParallelism(1)
val stream: DataStream[String] = env.socketTextStream("127.0.0.1",8080)
val textWithTsDstream: DataStream[(String, Long, Int)] = stream.map(text => {
val arr: Array[String] = text.split(" ")
(arr(0), arr(1).toLong, 1)
})
val textWithEventTimeDstream: DataStream[(String, Long, Int)] = textWithTsDstream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[(String, Long, Int)]() {
override def extractTimestamp(t: (String, Long, Int)) = t._2
})
val textKeyStream = textWithEventTimeDstream.keyBy(0)
textKeyStream.print("textKey:")
val windowStream = textKeyStream.window(TumblingEventTimeWindows.of(Time.seconds(2)))
val groupDstream: DataStream[mutable.HashSet[Long]] =
windowStream.fold(new mutable.HashSet[Long]()) {
case(set, (key, ts, count))=>
set += ts
}
groupDstream.print("window::::").setParallelism(1)
env.execute()
}
}
结果是按照Event Time的时间窗口计算得出的,而无关系统的时间(包括输入的快慢)。
4.2 滑动窗口(SlidingEventTimeWindows)
package com.flink.scala
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.{
DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.assigners.{
SlidingEventTimeWindows}
import org.apache.flink.streaming.api.windowing.time.Time
import scala.collection.mutable
object SlidingEventTimeWindowsTest {
def main(args: Array[String]): Unit = {
val env = StreamExecutionEnvironment.getExecutionEnvironment
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
env.setParallelism(1)
val stream: DataStream[String] = env.socketTextStream("127.0.0.1",8080)
val textWithTsDstream: DataStream[(String, Long, Int)] = stream.map(text => {
val arr: Array[String] = text.split(" ")
(arr(0), arr(1).toLong, 1)
})
val textWithEventTimeDstream: DataStream[(String, Long, Int)] = textWithTsDstream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[(String, Long, Int)]() {
override def extractTimestamp(t: (String, Long, Int)) = t._2
})
val textKeyStream = textWithEventTimeDstream.keyBy(0)
textKeyStream.print("textKey:")
val windowStream = textKeyStream.window(SlidingEventTimeWindows.of(Time.seconds(2),Time.milliseconds(500)))
val groupDstream: DataStream[mutable.HashSet[Long]] =
windowStream.fold(new mutable.HashSet[Long]()) {
case(set, (key, ts, count))=>
set += ts
}
groupDstream.print("window::::").setParallelism(1)
env.execute()
}
}
4.3 会话窗口(EventTimeSessionWindows)
相邻两次数据的 EventTime 的时间差超过指定的时间间隔就会触发执行。如果加入 Watermark,会在符合窗口触发的情况下进行延迟。到达延迟水位再进行窗口触发。
package com.flink.scala
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.{
DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.assigners.{
EventTimeSessionWindows, SlidingEventTimeWindows, TumblingEventTimeWindows}
import org.apache.flink.streaming.api.windowing.time.Time
object EventTimeSessionWindowsTest {
def main(args: Array[String]): Unit = {
val env = StreamExecutionEnvironment.getExecutionEnvironment
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
env.setParallelism(1)
val stream: DataStream[String] = env.socketTextStream("127.0.0.1",8080)
val textWithTsDstream: DataStream[(String, Long, Int)] = stream.map(text => {
val arr: Array[String] = text.split(" ")
(arr(0), arr(1).toLong, 1)
})
val textWithEventTimeDstream: DataStream[(String, Long, Int)] = textWithTsDstream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[(String, Long, Int)]() {
override def extractTimestamp(t: (String, Long, Int)) = t._2
})
val textKeyStream = textWithEventTimeDstream.keyBy(0)
textKeyStream.print("textKey:")
val windowStream = textKeyStream.window(EventTimeSessionWindows.withGap(Time.milliseconds(500)))
windowStream.reduce((text1,text2) =>
(text1._1, 0L, text1._3 + text2._3)
).map(_._3).print("windows:::").setParallelism(1)
env.execute()
}
}