Flink的处理机制以及侧输出应用

Posted 2023-03-26

参考技术A

Flink EventTime和Watermark
https://www.jianshu.com/p/5e735b63fb5b

Flink只要不用时间窗口函数，就是基于事件处理，对于事件驱动的任务，我们需要关心的点，尤其是存在shuffle和聚合的时候：
<1> 是否存在数据倾斜
<2> 是否会存在某些节点状态过大 （例如使用状态时，不配置过期时间，那么状态会一直缓存，就会导致内容一直增加，带来gc等问题）

基于窗口操作。

watermark机制，就可以理解为驱动flink基于事件时间处理的机制

一旦允许了元素滞后，那么滞后元素在截止时间前到达后也会触发计算并输出结果。
注意：滞后元素触发计算应该视为之前计算的更新结果，也即是相同计算会有多个结果。根据你的应用逻辑，要考虑是否会重复计算或者进行去重。

就是数据什么时候计算并落地。
Triggers定义了何时开始使用窗口计算函数计算窗口。每个WindowAssigner都会有一个默认的Trigger。如果，默认的Trigger不能满足你的需求，你可以指定一个自定义的trigger()。

抽象类Trigger：

onElement():进入窗口的每个元素都会调用该方法。
onEventTime():事件时间timer触发的时候被调用。
onProcessingTime():处理时间timer触发的时候会被调用。
onMerge():有状态的触发器相关，并在它们相应的窗口合并时合并两个触发器的状态，例如使用会话窗口。
clear():该方法主要是执行窗口的删除操作。

1).前三方法决定着如何通过返回一个TriggerResult来操作输入事件。
CONTINUE:什么都不做。
FIRE:触发计算。
PURE:清除窗口的元素。
FIRE_AND_PURE:触发计算和清除窗口元素。

2). 这些方法中的任何一个都可用于为将来的操作注册处理或事件时间计时器

所有基于事件时间的窗口分配器都用EventTimeTrigger作为默认触发器。该触发器会在watermark达到窗口的截止时间（window.maxTimestamp()）时直接触发计算输出。

注意:GlobalWindows的默认触发器是NeverTrigger，这个是不会触发计算的。因此，你需要为GlobalWindows自己定义一个触发器。

注意：通过调用trigger()方法指定触发器，就可以覆盖掉默认的触发器。例如，如果把TumblingEventTimeWindows的触发器指定为CountTrigger，那么就不会在根据时间进度触发窗口计算了。如果想同时支持时间和计数进行计算触发，那么就需要你自定义触发器了。

trigger()方法====》
在DataStream调用过窗口函数后，返回WindowedStream，就可以调用trigger()方法，指定新的Trigger替代默认Trigger。

窗口生命周期
简单来说，窗口创建于属于该窗口的第一个元素到达，结束于事件时间或者处理时间达到了窗口的结尾时间加上用户自定义的允许延迟时间。

分流：使用split函数；
异常值捕获：（比如空值，字段缺失数据，异常大值等），滞后数据处理等可以用sideoutput。

场景举例：
比如现在有一篇文章吧，单词长度不一，但是我们想对单词长度小于5的单词进行wordcount操作，同时又想记录下来哪些单词的长度大于了5，那么我们该如何做呢？

普遍的做法是：
datastream.filter(word.length>=5); //获取不统计的单词，也即是单词长度大于等于5。

datastream.filter(word.length <5);// 获取需要进行wordcount的单词。

这样数据，然后每次筛选都要保留整个流，然后遍历整个流，显然很浪费性能，假如能够在一个流了多次输出就好了， flink的侧输出提供了这个功能，侧输出的输出(sideoutput)类型可以与主流不同，可以有多个侧输出(sideoutput)，每个侧输出不同的类型。

在使用侧输出的时候需要先定义OutputTag
如:

OutputTag有两个构造函数，上面例子构造函数只有一个id参数，还有一个构造函数包括两个参数，id，TypeInformation信息。

要使用侧输出， 在处理数据的时候除了要定义相应类型的OutputTag外，还要使用特定的函数，主要是有四个 ：
ProcessFunction、CoProcessFunction、
ProcessWindowFunction、ProcessAllWindowFunction
1.8之后添加了 KeyedProcessFunction

这里滞后数据的类型应该需要与DataStream中的类型保持一致。