DataStream API编程模型
1.【Flink-Scala】DataStream编程模型之 数据源、数据转换、数据输出
2.【Flink-scala】DataStream编程模型之 窗口的划分-时间概念-窗口计算程序
3.【Flink-scala】DataStream编程模型之 窗口计算-触发器-驱逐器
文章目录
- DataStream API编程模型
- 前言
- 一、水位线
- 1.1 水位线
- 1.1.1 概念
- 1.1.2 水位线如何发挥作用呢?
- 1.1.3 水位线原理
- 1.1.3.1 消息正常到达系统的时间
- 1.1.3.2消息延迟到达系统时的情况
- 1.1.3.3 采用事件时间时的情况
- 1.1.3.4 引入水位线机制的情况
- 1.1.4 水位线的设置方法
- 1.1.4.1水位线生成策略--固定延迟生成水位线
- 1.1.4.2 水位线生成策略-单调递增生成水位线
- 1.1.4.3 自动义生成水位线策略
- 总结
前言
本小节学习水位线和延迟数据处理,再学习状态编程,水位线和延迟数据处理关联性强一点,如果篇幅太长,我就再开一小节写。
开始吧!
水位线,这和实际生活中河流/水库到达哪个水位线就要有什么问题一样,就是达到水位线后做什么处理。
(我这样想:河流里或者是水龙头的水是水流,把水换成数据就是datastream[数据流]。水位线这个概念也能扯上点关系)
一、水位线
1.1 水位线
1.1.1 概念
水位线是一种衡量事件时间进展的机制,它是数据本身的一个隐藏属性,本质上就是一个时间戳。
水位线是配合事件时间来使用的,通常基于事件时间的数据,自身都包含一个水位线用于处理乱序事件。
使用处理时间来处理事件时不会有延迟,因此也不需要水位线,所以水位线只出现在事件时间窗口。
正确地处理乱序事件,通常是结合窗口和水位线这两种机制来实现的。
1.1.2 水位线如何发挥作用呢?
在流处理过程中,从事件产生,到流经数据源,再到流经算子,中间是有一个过程和时间的。
虽然大部分情况下,流到算子的数据都是按照事件产生的时间顺序到达的,但是也不排除由于网络、系统等原因,导致乱序的产生和迟到数据。
但是对于迟到数据而言,我们又不能无限期地等下去,必须要有个机制来保证在经过一个特定的时间后,必须触发窗口去进行计算。
在进行窗口计算时,使用接入时间或处理时间的消息,都是以系统的墙上时间(比如现在是8:50我写的博客,那么这个事件就是8:50,就是生成时间)为准,因此事件都是按序到达的。但在实际应用中,由于网络或者系统等外部因素影响,事件数据往往不能及时到达Flink系统,从而造成数据乱序到达或者延迟到达等问题。针对这两个问题Flink主要采用以水位线为核心的机制来应对。
此时就是水位线发挥作用了,它表示当达到水位线后,在水位线之前的数据已经全部到达(即使后面还有延迟的数据),系统可以触发相应的窗口计算。
只有水位线越过窗口对应的结束时间,窗口才会关闭和进行计算。
一般而言,只有以下两个条件同时成立,才会触发窗口计算:
(1)条件T1:水位线时间 >= 窗口结束时间;
(2)条件T2:在[窗口开始时间,窗口结束时间)中有数据存在。
理想情况下,水位线应该与处理时间一致,并且处理时间与事件时间只相差常数时间甚至为零。
当水位线与处理时间完全重合时,就意味着消息产生后马上被处理,不存在消息迟到的情况。
然而,由于网络拥塞或系统原因,消息常常存在迟到的情况,
因此,在设置水位线时,总是考虑一定的延时,从而给予迟到的数据一些机会。具体的延迟大小根据水位线实现方式的不同而也有所差别
1.1.3 水位线原理
1.1.3.1 消息正常到达系统的时间
window1[5s-15s]
window2 [10s-20s]
window3[15s-25s]
现在假设有一个单词数据流,需要采用基于处理时间的滑动窗口进行实时的词频统计,滑动窗口大小为10s,滑动步长为5s。
假设数据源分别在第12秒、第12秒和第17秒的时候,生成3条内容为单词“a”的消息,这些消息将进入窗口中。
时间<15s, 2条数据
时间>15s,1条数据
5-15s,3条数据
每个窗口提交后,最后统计值分别是 (a, 2),(a, 3) 和 (a, 1)
1.1.3.2消息延迟到达系统时的情况
正常是12-17s来了3条数据,现在开始有迟到数据。
假设在12s时候出现一条迟到6s的数据(18sde 数据),这条延迟的消息会落入 window2 [10s-20s] 和 window3[15s-25s]。
窗口提交后,最后统计值将分别是 (a, 1),(a, 3) 和 (a, 2)。(正常应该是(a, 2),(a, 3) 和 (a, 1))
可看出这条延迟的消息没有对window2 [10s-20s]的计算结果造成影响,但却影响了window1[5s-15s]和 window3[15s-25s]的计算结果,导致二者计算结果出错。
因为当这条消息在第18秒到达时,window1[5s-15s]计算已结束,这条消息不会被统计到window1[5s-15s]中,而会落入window3[15s-25s],导致被统计window3[15s-25s]
1.1.3.3 采用事件时间时的情况
采用事件时间,则当系统时间行进到第18秒时,这条迟到了6秒的消息会落入 window2 [10s-20s],
因为这条消息的事件生成时间是第12秒,所以就应该属于window1[5s-15s]和window2 [10s-20s],
但是在第18秒时,window1[5s-15s]已经关闭,所以,这条延迟的消息只会落入 window2 [10s-20s]。
最终,三个窗口的计算结果是(a,1),(a, 3) 和 (a, 1),也就是说,window2[10s-20s]和 window3[15s-25s]提交了正确的结果,但是 window1[5s-15s]的结果还是错误的
1.1.3.4 引入水位线机制的情况
就本例而言,水位线本质上就是告诉Flink一条消息可以延迟多久,
因此,这里让水位线等于系统当前时间减去5秒。由于只有水位线越过窗口对应的结束时间,窗口才会关闭和进行计算,
因此,第1个窗口window1[5s-15s]将会在第20秒的时候进行计算,第2个窗口window2[10s-20s]将会在第25秒的时候进行计算,第3个窗口window3[15s-25s]将会在第30秒的时候进行计算。
当系统时间行进到第18秒时,这条迟到了6秒的消息会落入window1[5s-15s])和 window2 [10s-20s],因为这条消息的事件生成时间是第12秒,所以就应该属于window1[5s-15s]和window2 [10s-20s]。
最终,三个窗口提交正确结果,即(a, 2),(a, 3) 和 (a, 1)
1.1.4 水位线的设置方法
水位线事关事件时间,那么就需要知道事件时间戳。
就必须为数据流中的每个元素分配一个时间戳。
在Flink系统中,分配时间戳和生成水位线这两个工作是同时进行的,前者是由TimestampAssigner来实现的,后者则是由WatermarkGenerator来实现的。
当我们构建了一个DataStream之后,可以使用assignTimestampsAndWatermarks方法来分配时间戳和生成水位线,调用该方法时,需要传入一个WatermarkStrategy对象,语法如下:
DataStream.assignTimestampsAndWatermarks(WatermarkStrategy<T>)
一般情况下,Flink要求WatermarkStrategy对象中同时包含了TimestampAssigner对象和WatermarkGenerator对象。
WatermarkStrategy是一个接口,提供了很多静态的方法,对于一些常用的水位线生成策略,我们不需要去实现这个接口,可以直接调用静态方法来生成水位线。
或者,我们也可以通过实现WatermarkStrategy接口中的createWatermarkGenerator方法和createTimestampAssigner方法,来自定义水位线策略。
说到底就是两个,分配时间戳,生成水位线(有的地方叫水印)。
1.1.4.1水位线生成策略–固定延迟生成水位线
固定延迟生成水位线的语法如下:
WatermarkStrategy.forBoundedOutOfOrderness(Duration maxOutOfOrderness)
比如,现在要实现一个延迟3秒的固定延迟水位线,并从消息中获取时间戳,具体语句如下:
val dataStream = ......
dataStream.assignTimestampsAndWatermarks(
WatermarkStrategy
.forBoundedOutOfOrderness[StockPrice](Duration.ofSeconds(3))//这里延迟3s
.withTimestampAssigner(new SerializableTimestampAssigner[StockPrice] {
override def extractTimestamp(element: StockPrice, recordTimestamp: Long): Long = element.timeStamp
//分配时间戳
}
)
)
使用的是这个方法forBoundedOutOfOrderness
1.1.4.2 水位线生成策略-单调递增生成水位线
单调递增生成水位线是通过WatermarkStrategy接口的静态方法forMonotonousTimestamps提供的,语法如下:
WatermarkStrategy.forMonotonousTimestamps()
学习单词:
在程序中按照如下方式使用:
val dataStream = ......
dataStream.assignTimestampsAndWatermarks(
WatermarkStrategy
.forMonotonousTimestamps()
.withTimestampAssigner(new SerializableTimestampAssigner[StockPrice] {
override def extractTimestamp(element: StockPrice, recordTimestamp: Long): Long = element.timeStamp
}
)
)
1.1.4.3 自动义生成水位线策略
自定义肯定就是实现某个接口的什么方法啦,之前就说过
水位线设置就两个:分配时间戳,生成水位线
这里我们只需要实现WatermarkStrategy接口中的createWatermarkGenerator
方法和createTimestampAssigner
方法就可以了。
水位线策略:
createWatermarkGenerator方法需要返回一个WatermarkGenerator对象。
WatermarkGenerator是一个接口,需要实现这个接口里面的onEvent方法和onPeriodicEmit方法:
(1)onEvent:数据流中的每个元素(或事件)到达以后,都会调用这个方法,如果我们想依赖每个元素生成一个水位线,然后发射到下游,就可以实现这个方法。
(2)onPeriodicEmit:当数据量比较大的时候,为每个元素都生成一个水位线,会影响系统性能,所以Flink还提供了一个周期性生成水位线的方法。这个水位线的生成周期的设置方法是:env.getConfig.setAutoWatermarkInterval(5000L),其中5000L是间隔时间,可以由用户自定义。
在自定义水位线生成策略时,Flink提供了两种不同的方式:
1.定期水位线:在这种机制中,系统会通过onEvent方法对系统中到达的事件进行监控,然后,在系统调用onPeriodicEmit方法时,生成一个水位线。(两个方法都使用)
2.标点水位线:在这种机制中,系统会通过onEvent方法对系统中到达的事件进行监控,并等待具有特定标记的事件到达,一旦监测到特定事件到达,就立即生成一个水位线。通常,这种机制不会调用onPeriodicEmit方法来生成一个水位线。(只使用一个方法)
代码:
import java.text.SimpleDateFormat
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, TimestampAssigner, TimestampAssignerSupplier, Watermark, WatermarkGenerator, WatermarkGeneratorSupplier, WatermarkOutput, WatermarkStrategy}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
case class StockPrice(stockId:String,timeStamp:Long,price:Double)
object WatermarkTest {
def main(args: Array[String]): Unit = {
//设定执行环境
val env = StreamExecutionEnvironment.getExecutionEnvironment
//设定时间特性为事件时间
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
//设定程序并行度
env.setParallelism(1)
//创建数据源
val source = env.socketTextStream("localhost", 9999)
//指定针对数据流的转换操作逻辑
val stockDataStream = source
.map(s => s.split(","))
.map(s=>StockPrice(s(0).toString,s(1).toLong,s(2).toDouble))
//为数据流分配时间戳和水位线
val watermarkDataStream = stockDataStream.assignTimestampsAndWatermarks(new MyWatermarkStrategy)
//执行窗口计算
val sumStream = watermarkDataStream
.keyBy("stockId")
.window(TumblingEventTimeWindows.of(Time.seconds(3)))
.reduce((s1, s2) => StockPrice(s1.stockId,s1.timeStamp, s1.price + s2.price))
//打印输出
sumStream.print("output")
//指定名称并触发流计算
env.execute("WatermarkTest")
}
//指定水位线生成策略
class MyWatermarkStrategy extends WatermarkStrategy[StockPrice] {
override def createTimestampAssigner(context:TimestampAssignerSupplier.Context):TimestampAssigner[StockPrice]={
new SerializableTimestampAssigner[StockPrice] {
override def extractTimestamp(element: StockPrice, recordTimestamp: Long): Long = {
element.timeStamp //从到达消息中提取时间戳
}
}
}
override def createWatermarkGenerator(context:WatermarkGeneratorSupplier.Context): WatermarkGenerator[StockPrice] ={
new WatermarkGenerator[StockPrice](){
val maxOutOfOrderness = 10000L //设定最大延迟为10秒
var currentMaxTimestamp: Long = 0L
var a: Watermark = null
val format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS")
override def onEvent(element: StockPrice, eventTimestamp: Long, output:WatermarkOutput): Unit = {
currentMaxTimestamp = Math.max(eventTimestamp, currentMaxTimestamp)
a = new Watermark(currentMaxTimestamp - maxOutOfOrderness)
output.emitWatermark(a)
println("timestamp:" + element.stockId + "," + element.timeStamp + "|" + format.format(element.timeStamp) + "," + currentMaxTimestamp + "|" + format.format(currentMaxTimestamp) + "," + a.toString)
}
override def onPeriodicEmit(output:WatermarkOutput): Unit = {
// 没有使用周期性发送水印,因此这里没有执行任何操作
}
}
}
}
}
输入:
s1 stock_1,1602031567000,8.14
s2 stock_1,1602031571000,8.23
s3 stock_1,1602031577000,8.24
s4 stock_1,1602031578000,8.87
s5 stock_1,1602031579000,8.55
s6 stock_1,1602031581000,8.43
s7 stock_1,1602031582000,8.78
然后,在日志终端内,就可以看到如下输出信息:
timestamp:stock_1,1602031567000|2020-10-07 08:46:07.000,1602031567000|2020-10-07 08:46:07.000,Watermark @ 1602031557000 (2020-10-07 08:45:57.000)
timestamp:stock_1,1602031571000|2020-10-07 08:46:11.000,1602031571000|2020-10-07 08:46:11.000,Watermark @ 1602031561000 (2020-10-07 08:46:01.000)
timestamp:stock_1,1602031577000|2020-10-07 08:46:17.000,1602031577000|2020-10-07 08:46:17.000,Watermark @ 1602031567000 (2020-10-07 08:46:07.000)
timestamp:stock_1,1602031578000|2020-10-07 08:46:18.000,1602031578000|2020-10-07 08:46:18.000,Watermark @ 1602031568000 (2020-10-07 08:46:08.000)
timestamp:stock_1,1602031579000|2020-10-07 08:46:19.000,1602031579000|2020-10-07 08:46:19.000,Watermark @ 1602031569000 (2020-10-07 08:46:09.000)
output> StockPrice(stock_1,1602031567000,8.14)
timestamp:stock_1,1602031581000|2020-10-07 08:46:21.000,1602031581000|2020-10-07 08:46:21.000,Watermark @ 1602031571000 (2020-10-07 08:46:11.000)
timestamp:stock_1,1602031582000|2020-10-07 08:46:22.000,1602031582000|2020-10-07 08:46:22.000,Watermark @ 1602031572000 (2020-10-07 08:46:12.000)
output> StockPrice(stock_1,1602031571000,8.23)
为了正确理解水位线的工作原理,下面我们详细解释每个事件到达后水位线的变化情况、各个窗口中的事件分布情况以及窗口触发计算的情况。关于窗口计算,这里要再次强调,只有以下两个条件同时成立,才会触发窗口计算:
(1)条件T1:水位线时间 >= 窗口结束时间;
(2)条件T2:在[窗口开始时间,窗口结束时间)中有数据存在。
1.s1事件到达后
事件s1到达系统以后的水位线的变化情况,可以看出,当前的水位线已经到达了1602031557000(2020-10-07 08:45:57.000)。
s1到达后各个窗口包含事件的情况
水位线是在增长的,在那么增长的呢?
这是我截取上面 的部分代码。最大延迟10s,就是本次到的事件最大时间戳-10s,即为水位线。对应下代码:a = new Watermark(currentMaxTimestamp - maxOutOfOrderness)
val maxOutOfOrderness = 10000L //设定最大延迟为10秒
var currentMaxTimestamp: Long = 0L
var a: Watermark = null
val format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS")
override def onEvent(element: StockPrice, eventTimestamp: Long, output:WatermarkOutput): Unit = {
currentMaxTimestamp = Math.max(eventTimestamp, currentMaxTimestamp)
a = new Watermark(currentMaxTimestamp - maxOutOfOrderness)
水位线增长:每次有新事件到达时,都会检查并更新currentMaxTimestamp,然后根据这个值减去maxOutOfOrderness来生成新的水位线
2.当事件s2到达以后
s2到达系统以后的水位线的变化情况,可以看出,当前的水位线已经到达了1602031561000(2020-10-07 08:46:01.000)。
s2到达以后各个窗口内包含的事件的情况。
3.当事件s3到达以后
事件s3到达系统以后的水位线的变化情况,可以看出,当前的水位线已经到达了1602031567000(2020-10-07 08:46:07.000)。
s3到达以后各个窗口内包含的事件的情况。
4.当事件s4到达以后
事件s4到达系统以后的水位线的变化情况,可以看出,当前的水位线已经到达了1602031568000(2020-10-07 08:46:08.000)。
s4到达以后各个窗口内包含的事件的情况。
回顾一下:
触发窗口计算:
(1)条件T1:水位线时间 >= 窗口结束时间;
(2)条件T2:在[窗口开始时间,窗口结束时间)中有数据存在。
看到水位线事件8:46:08,窗口结束事件是09,那么此时还没有大于等于。
继续
5.当事件s5到达以后
事件s5到达系统以后的水位线的变化情况,可以看出,当前的水位线已经到达了1602031569000(2020-10-07 08:46:09.000)。
当当当,注意啦,看看触发窗口计算条件。
s5到达以后各个窗口内包含的事件的情况。
8:46:09水位线已经大于等于w1窗口结束时间啦,条件1满足,且窗口有数据,条件2满足,w1开始计算!!!
6.当事件s6到达以后
事件s6到达系统以后的水位线的变化情况,可以看出,当前的水位线已经到达了1602031571000(2020-10-07 08:46:11.000)。
s6到达以后各个窗口内包含的事件的情况:
此时再看看条件满足否?
7.当事件s7到达以后
事件s7到达系统以后的水位线的变化情况,可以看出,当前的水位线已经到达了1602031572000(2020-10-07 08:46:12.000)。
s7到达以后各个窗口内包含的事件的情况。
当当当,又注意啦,看看是否满足条件?
满足条件,触发计算,窗口2 计算完成!!
总结
没有想到水位线写了这么多,延迟数据处理还没有写,本小节主要学习水位线的原理和设置方法。
其中自定义的水位线生成策略稍显麻烦,代码需要着重分析。下一小节该写延迟数据处理了。