三、窗口

 

1、窗口的介绍

 

（1）含义

 

将无限的流式数据切割为有限块处理，以便于聚合等操作

 

（2）图解

 

 

 

2、窗口的分类

 

（1）按性质分

 

Flink 支持三种划分窗口的方式，time、count和会话窗口（Session Windows）：session间隔定义了非活跃周期的长度，一段时间没有接收到新数据就会生成新的窗口。如果根据时间划分窗口，那么它就是一个time-window（时间窗口）；如果根据数据划分窗口，那么它就是一个count-window（数量窗口）。一段时间没有接收到新数据就会生成新的窗口，则为会话窗口。

 

（2）按类型分

 

窗口又可以分为滚动窗口（Tumbling Window）、滑动窗口（Sliding Window）和会话窗口（Session Window）。

 

滚动窗口无重叠数据，而滑动窗口有重叠数据。

 

3、窗口API

 

窗口包含两个重要属性（size窗口大小和interval间隔-多久统计一次），如果size=interval，那么就会形成tumbling-window(无重叠数据) 如果size>interval，那么就会形成sliding-window(有重叠数据) 如果size< interval，那么窗口将会丢失数据。

 

例如：每5秒钟，统计过去3秒的通过路口汽车的数据，将会漏掉2秒钟的数据

 

组合后可以形成下列四种窗口

 

time-tumbling-window 无重叠数据的时间窗口，设置方式举例：timeWindow(Time.seconds(5))

 

time-sliding-window 有重叠数据的时间窗口，设置方式举例：timeWindow(Time.seconds(5), Time.seconds(3))

 

count-tumbling-window无重叠数据的数量窗口，设置方式举例：countWindow(5)

 

count-sliding-window 有重叠数据的数量窗口，设置方式举例：countWindow(5,3)

 

4、API补充

 

windowfunction：增量聚合（每次到来都计算）、全窗口函数（全部到来再遍历）

 

其它可选API：.trigger()——触发器、.evitor()——移除器、.allowedLateness()——允许处理迟到的数据、.getSideOutput() —— 获取侧输出流

 

四、时间语义与水印

 

1、时间语义分类

 

流式数据处理的时间可以分为事件时间，进入时间和处理时间三种

 

Event Time：事件的创建时间，消息本身携带

 

Ingestion Time：进入时间，以client客户端时间为准

 

Processing Time：处理时间（默认的时间属性），以服务端时间为准

 

通常根据日志的生成时间（Event Time）进行统计

 

引入时间语义：

 

val env = StreamExecutionEnvironment.getExecutionEnvironment // 从调用时刻开始给env创建的每一个stream追加时间特征 env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

 

2、Watermark水印

 

（1）含义

 

Flink 为了处理 EventTime 窗口延迟计算提出的一种机制，本质是一种时间戳。每条消息都有一个事件时间和一个水印时间（计算得出，例如maxEventTime-t）。

 

通常与Window一起处理乱序事件。由于网络延迟等原因，不能无限期的等下去，保证特定时间后，必须触发窗口进行计算。

 

（2）实现原理

 

添加水印后，窗口会等5秒，再执行计算。若超过5秒，则舍弃。

 

窗口执行计算时间由水印时间来触发（而非窗口结束时间），当接收到消息的watermark >= endtime时，触发窗口的计算

 

（3）具体操作

 

实现TimestampAssigner接口，实现根据事件时间计算水印时间

 

五、ProcessFunctionAPI（底层API）

 

1、含义

 

是flink的底层转换算子，通过这些底层转换算子可以访问数据的时间戳、watermark以及注册定时事件等，还可以输出特定的一些事件，例如超时事件等。

 

2、组成

 

ProcessFunction

 

KeyedProcessFunction

 

CoProcessFunction

 

ProcessJoinFunction

 

BroadcastProcessFunction

 

KeyedBroadcastProcessFunction

 

ProcessWindowFunction

 

ProcessAllWindowFunction

 

3、调用方式

 

inputData.flatMap(new MySpliter())     .process(new KeyedProcessFunction<String, Sensor, String>() {})

 

 

 

所有的Process Function都实现了RichFunction接口，都有open()、close()和getRuntimeContext()等方法

 

4、KeyedProcessFunction

 

按照key对元素进行处理，额外提供下列两个方法

 

processElement(Sensor sensor, Context context, Collector collector), 流中的每个元素都会调用此方法，调用结果将会放在Collector数据类型中输出；context包含上下文信息，包括当前数据的时间戳、key以及时间服务，还可以将数据放到侧输出流

 

onTimer(long timestamp, OnTimerContext ctx, Collector out)是一个回调函数。当定时器触发时调用。timestamp是定时器的触发时间戳，ctx上下文信息，out收集输出信息

 

5、TimerService和定时器对象

 

（1）含义及组成

 

TimerService是Context上下文和OnTimerContext 的对象，均包含下列方法：

 

//返回当前处理时间 long currentProcessingTime(); //返回当前水位线 long currentWatermark(); //注册当前key的process time定时器，process time到达定时时间时触发timer void registerProcessingTimeTimer(long var1); //注册当前key的event time定时器，当watermark大于等于定时时间时触发timer void registerEventTimeTimer(long var1); //删除指定时间戳的process time定时器 void deleteProcessingTimeTimer(long var1); //删除指定时间戳的event time定时器 void deleteEventTimeTimer(long var1);

 

（3）举例：温度传感器

 

传感器温度在1秒内持续升高则发出报警信息。

 

.process(new KeyedProcessFunction<String, Sensor, String>() 实现processElement()方法负责根据值注册和清除定时器 实现onTimer()方法用于发出报警【回调函数】

6、侧输出流（SideOutput）

 

（1）含义

 

默认算子单一输出，用侧输出流可以产生多条不同数据类型的流

 

每个输出流可以定义为定义为OutputTag[X]对象，并通过Context对象发射到指定事件或对象

 

（2）使用方式

 

val monitoredReadings: DataStream[SensorReading] = readings   .process(new FreezingMonitor) monitoredReadings   .getSideOutput(new OutputTag[String]("freezing-alarms"))   .print() //输出指定的流信息 readings.print() //输出全部的流信息

 

（3）FreezingMonitor函数具体实现

 

class FreezingMonitor extends ProcessFunction[SensorReading, SensorReading] {   // 定义一个侧输出标签   lazy val freezingAlarmOutput: OutputTag[String] =     new OutputTag[String]("freezing-alarms")     override def processElement(r: SensorReading,                               ctx: ProcessFunction[SensorReading, SensorReading]#Context,                               out: Collector[SensorReading]): Unit = {     // 温度在32F以下时，输出警告信息     if (r.temperature < 32.0) {       ctx.output(freezingAlarmOutput, s"Freezing Alarm for ${r.id}")     }     // 所有数据直接常规输出到主流     out.collect(r)   } }

 

7、CoProcessFunction

 

（1）含义

 

使用CoProcessFunction可以合并两条流，根据id将两个流中的数据组合

 

（2）实现

 

提供了操作每一个输入流的方法:processElement1()和processElement2()

 

第一个流ValueState<String>的值不为空，则在第二个流中合并

 

// 流2的处理逻辑与流1的处理逻辑类似     @Override     public void processElement2(String value, Context ctx, Collector<Tuple2<String, String>> out) throws Exception {         String value1 = state1.value();         if (value1 != null) {             out.collect(Tuple2.of(value1, value));             state1.clear();             ctx.timerService().deleteEventTimeTimer(timeState.value());             timeState.clear();         } else {             state2.update(value);             long time = 1111L + 60000;             timeState.update(time);             ctx.timerService().registerEventTimeTimer(time);         } } //在定时器中将value不为空的tag进行输出