1 窗口

1.1 说说Flink窗口，以及划分机制

窗口概念：将无界流的数据，按时间区间，划分成多份数据，分别进行统计(聚合)

划分机制：

Flink支持两种划分窗口的方式(time和count)，第一种，按时间驱动进行划分、另一种按数据驱动进行划分。

1、按时间驱动Time Window 划分可以分为 滚动窗口 Tumbling Window 和滑动窗口 Sliding Window。

2、按数据驱动Count Window也可以划分为滚动窗口 Tumbling Window 和滑动窗口 Sliding Window。

3、Flink支持窗口的两个重要属性(窗口长度size和滑动间隔interval)，通过窗口长度和滑动间隔来区分滚动窗口和滑动窗口。

如果size=interval,那么就会形成tumbling-window(无重叠数据)--滚动窗口如果size(1min)>interval（30s）,那么就会形成sliding-window(有重叠数据)--滑动窗口

通过组合可以得出四种基本窗口：

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

 （2）time-sliding-window  有重叠数据的时间窗口，设置方式举例：timeWindow(Time.seconds(10), Time.seconds(5))---基于时间的滑动窗口

 （3）count-tumbling-window无重叠数据的数量窗口，设置方式举例：countWindow(5)---基于数量的滚动窗口

 （4）count-sliding-window 有重叠数据的数量窗口，设置方式举例：countWindow(10,5)---基于数量的滑动窗口

Flink中还支持一个特殊的窗口:会话窗口SessionWindows

session窗口分配器通过session活动来对元素进行分组，session窗口跟滚动窗口和滑动窗口相比，不会有重叠和固定的开始时间和结束时间的情况session窗口在一个固定的时间周期内不再收到元素，即非活动间隔产生，那个这个窗口就会关闭。一个session窗口通过一个session间隔来配置，这个session间隔定义了非活跃周期的长度，当这个非活跃周期产生，那么当前的session将关闭并且后续的元素将被分配到新的session窗口中去,如下图所示：

 1.2 介绍下Flink的窗口机制以及各组件之间是如何相互工作的

 WindowAssigner

1、窗口算子负责处理窗口，数据流源源不断地进入算子（window operator）时，每一个到达的元素首先会被交给 WindowAssigner。WindowAssigner 会决定元素被放到哪个或哪些窗口（window），可能会创建新窗口。因为一个元素可以被放入多个窗口中（个人理解是滑动窗口，滚动窗口不会有此现象），所以同时存在多个窗口是可能的。注意，Window本身只是一个ID标识符，其内部可能存储了一些元数据，如TimeWindow中有开始和结束时间，但是并不会存储窗口中的元素。窗口中的元素实际存储在 Key/Value State 中，key为Window，value为元素集合（或聚合值）。为了保证窗口的容错性，该实现依赖了 Flink 的 State 机制。

WindowTrigger

2、每一个Window都拥有一个属于自己的 Trigger，Trigger上会有定时器，用来决定一个窗口何时能够被计算或清除。每当有元素加入到该窗口，或者之前注册的定时器超时了，那么Trigger都会被调用。Trigger的返回结果可以是 ：

（1）continue（继续、不做任何操作），

（2）Fire（触发计算，处理窗口数据），

（3）Purge（触发清理，移除窗口和窗口中的数据），

（4）Fire + purge（触发计算+清理，处理数据并移除窗口和窗口中的数据）。

当数据到来时，调用Trigger判断是否需要触发计算，如果调用结果只是Fire的话，那么会计算窗口并保留窗口原样，也就是说窗口中的数据不清理，等待下次Trigger fire的时候再次执行计算。窗口中的数据会被反复计算，直到触发结果清理。在清理之前，窗口和数据不会释放没所以窗口会一直占用内存。

Trigger 触发流程：

3、当Trigger Fire了，窗口中的元素集合就会交给Evictor（如果指定了的话）。Evictor 主要用来遍历窗口中的元素列表，并决定最先进入窗口的多少个元素需要被移除。剩余的元素会交给用户指定的函数进行窗口的计算。如果没有 Evictor 的话，窗口中的所有元素会一起交给函数进行计算。

计算函数

4、计算函数收到了窗口的元素（可能经过了 Evictor 的过滤），并计算出窗口的结果值，并发送给下游。窗口的结果值可以是一个也可以是多个。DataStream API 上可以接收不同类型的计算函数，包括预定义的sum(),min(),max()，还有 ReduceFunction，FoldFunction，还有WindowFunction。WindowFunction 是最通用的计算函数，其他的预定义的函数基本都是基于该函数实现的。5、Flink 对于一些聚合类的窗口计算（如sum,min）做了优化，因为聚合类的计算不需要将窗口中的所有数据都保存下来，只需要保存一个result值就可以了。每个进入窗口的元素都会执行一次聚合函数并修改result值。这样可以大大降低内存的消耗并提升性能。但是如果用户定义了 Evictor，则不会启用对聚合窗口的优化，因为 Evictor 需要遍历窗口中的所有元素，必须要将窗口中所有元素都存下来。

2 WATERMARK(TIME)

2.1 Flink的Time概念

在Flink的流式处理中，会涉及到时间的不同概念，主要分为三种时间机制，如下图所示：

- EventTime[事件时间]

 事件发生的时间，例如：点击网站上的某个链接的时间，每一条日志都会记录自己的生成时间。

如果以EventTime为基准来定义时间窗口那将形成EventTimeWindow,要求消息本身就应该携带EventTime

IngestionTime[摄入时间] 

数据进入Flink的时间，如某个Flink节点的sourceoperator接收到数据的时间，例如：某个source消费到kafka中的数据

如果以IngesingtTime为基准来定义时间窗口那将形成IngestingTimeWindow,以source的systemTime为准

ProcessingTime[处理时间] 

某个Flink节点执行某个operation的时间，例如：timeWindow处理数据时的系统时间，默认的时间属性就是Processing Time如果以ProcessingTime基准来定义时间窗口那将形成ProcessingTimeWindow，以operator的systemTime为准

在Flink的流式处理中，绝大部分的业务都会使用EventTime，一般只在EventTime无法使用时，才会被迫使用ProcessingTime或者IngestionTime。如果要使用EventTime，那么需要引入EventTime的时间属性。

2.3 API调用时，应该怎么使用？

final StreamExecutionEnvironment env  
    = StreamExecutionEnvironment.getExecutionEnvironrnent();
// 使用处理时间
env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime) ; 
// 使用摄入时间

env.setStrearnTimeCharacteristic(TimeCharacteristic.IngestionTime);

// 使用事件时间
env.setStrearnTimeCharacteristic(TimeCharacteri stic Eve~tTime);

2.4 迟到数据处理

一般处理数据延迟、消息乱序等问题，通过WaterMark水印来处理。

水印是用来解决数据延迟、数据乱序等问题，总结如下图所示：

水印就是一个时间戳（timestamp），Flink可以给数据流添加水印

        - 水印并不会影响原有Eventtime事件时间

        - 当数据流添加水印后，会按照水印时间来触发窗口计算,也就是说watermark水印是用来触发窗口计算的

        - 设置水印时间，会比事件时间小几秒钟,表示最大允许数据延迟达到多久

        - 水印时间 = 事件时间 - 允许延迟时间 (例如：10:09:57 =  10:10:00 - 3s )

2.5 WaterMark原理讲解一下？

2.6 WaterMark使用

2.7 如果数据延迟非常严重呢？只使用WaterMark可以处理吗？那应该怎么解决？

使用 WaterMark+ EventTimeWindow 机制可以在一定程度上解决数据乱序的问题，但是，WaterMark 水位线也不是万能的，在某些情况下，数据延迟会非常严重，即使通过Watermark + EventTimeWindow也无法等到数据全部进入窗口再进行处理，因为窗口触发计算后，对于延迟到达的本属于该窗口的数据，Flink默认会将这些延迟严重的数据进行丢弃那么如果想要让一定时间范围的延迟数据不会被丢弃，可以使用Allowed Lateness(允许迟到机制/侧道输出机制)设定一个允许延迟的时间和侧道输出对象来解决
即使用WaterMark + EventTimeWindow + Allowed Lateness方案（包含侧道输出），可以做到数据不丢失。

API调用

l  allowedLateness(lateness:Time)---设置允许延迟的时间该方法传入一个Time值，设置允许数据迟到的时间，这个时间和watermark中的时间概念不同。再来回顾一下，watermark=数据的事件时间-允许乱序时间值随着新数据的到来，watermark的值会更新为最新数据事件时间-允许乱序时间值，但是如果这时候来了一条历史数据，watermark值则不会更新。总的来说，watermark永远不会倒退它是为了能接收到尽可能多的乱序数据。那这里的Time值呢？主要是为了等待迟到的数据，如果属于该窗口的数据到来，仍会进行计算，后面会对计算方式仔细说明注意：该方法只针对于基于event-time的窗口 

l  sideOutputLateData(outputTag:OutputTag[T])--保存延迟数据

该方法是将迟来的数据保存至给定的outputTag参数，而OutputTag则是用来标记延迟数据的一个对象。l  DataStream.getSideOutput(tag:OutputTag[X])--获取延迟数据通过window等操作返回的DataStream调用该方法，传入标记延迟数据的对象来获取延迟的数据