Flink系列文档-(YY09)-Flink时间语义

1 三种时间语义

在实时流式计算中，"时间"是一个能影响计算结果的非常重要因素！

试想场景：每隔1分钟计算一次最近10分钟的活跃用户量：

①假设此刻的时间是13:10，要计算的活跃用户量时间段为：[ 13:00,13:10 )；

②有一条行为日志中记录的用户的行为时间是12:59，但到达flink计算程序时已是13:02；

那么，这个用户是否要纳入本次计算的结果中呢？看如何定义：

①如果时段 [13:00 , 13:10 ）定义的是用户行为的发生时间（数据中的业务时间），则不应纳入；

②如果时段 [13:00 , 13:10 ）定义的是计算时的时间，则应该纳入；

上面的问题是:数据的产生时间和数据的处理时间不一致

flink内部为了直观地统一计算时所用的时间标准，特制定了两种时间语义：

1. processing time   处理时间
2. event time  事件时间
3. Ingestion time 注入时间

时间语义主要影响 "窗口计算" ；

 2 两种时间语义

时间语义，是flink中用于时间推进和时间判断的机制；

时间推进和时间判断，以什么为标准，就产出了两种不同的时间语义；

1. 以 processing time为依据，则叫做处理时间语义
2. 以 event time为依据，则叫做事件时间语义

  时间语义的设计意义 

process(EventLog eventlog){
Long  eventTime  =  eventLog.getTimestamp();
Long  processTime = System.currentMillimise()
      // 用户完全可以自己根据需求中的时间定义来进行相应的计算
}

Flink为什么还要搞出一个  “事件时间语义”：时间按数据中的业务时间戳来推进！

主要是，实时流式计算中，有大量跟时间相关的统计需求，比如：时间窗口计算，定时器等，而这些需求，如果都让用户像上面的代码那样自己去进行判断、处理，那么它觉得自己的api不够强大！

所以，flink想在api的层面，将两类时间定义的计算需求进行api层面的统一，它才搞出这么一种”事件时间语义“，有了这种语义，那么，处理时间 和  事件时间，都可以看成 ”时间”

用户在不同时间定义下，要进行一个定时动作时，就不需要再像上面的代码那种去进行各种判断，而是一个统一的动作：  到 xxx时间，给我做个什么事！

process(EventLog eventlog，TimeStamp timestamp){

      // 不管需求是需要用哪种时间来计算，用户代码只需要看到一个timestamp了

}

代码中的timestamp到底是事件时间，还是处理时间，取决于环境中设置的“时间语义”

  处理时间（processing time）语义

Processing Time是指数据被Operator处理时所在机器的系统时间。

处理时间遵循客观世界中时间的特性：单调递增，恒定速度，永不停滞，永不回退；

  事件时间（event time）语义

Event Time是指在数据本身的业务时间（如用户行为日志中的用户行为时间戳）；

Event Time语义中，时间的推进完全由流入flink系统的数据来驱动：

数据中的业务时间推进到哪，flink就认为自己的时间推进到了哪

它可能停滞，也可能速度不恒定，但也一定是单调递增不可回退！

3 设置时间语义

1.12及以后，flink以event time作为默认的时间语义，在需要指定时间语义的相关操作（如时间窗口）时，可以通过显式的api来使用特定的时间语义；

上面我们介绍到 时间语义主要影响 "窗口计算" ；

API设置时间语义

keyedStream.window(SlidingEventTimeWindows.of(Time.seconds(5),Time.seconds(1)));

keyedStream.window(SlidingProcessingTimeWindows.of(Time.seconds(5),Time.seconds(1)));

keyedStream.window(TumblingEventTimeWindows.of(Time.seconds(5)));

keyedStream.window(TumblingProcessingTimeWindows.of(Time.seconds(5)));

 API中禁用时间语义

如果需要禁用event time机制，则可以通过设置watermark生成频率间隔来实现：

// 如果设置为0，则禁用了watermark的生成；从而失去了event time语义

ExecutionConfig.setAutoWatermarkInterval(long)；

提示：如果需要使用已过期的 ingestion time，可以通过设置恰当的watermark来实现；