在现实世界，数据是无界的。有可能产生乱序、延迟现象。例如用户在飞机上是飞行模式（离线模式），此时正在听spotify的歌曲，但是直到飞机降落才上线，此时数据的产生就是乱序的数据。而且由于经过kafka，由于并行处理的网络等原因，迟到的数据也是无处不在。
首先，第一步我们需要从kafka中消费数据。通过Flink内部的检查点机制，可以保证exactly once的处理，这仅仅需要提供几个kafka的参数：
sEnv.addSource(new FlinkKafkaConsumer09[Event](conf.topic(),
getSerializationSchema,
kafkaProperties(conf.kafkaBroker()))
)
之后，我们基于用户key，设置一个session window的gap，在同一个session window中的数据表示用户活跃的区间，假如gap的时间是15分钟：
.keyBy(_.userId)
.window(EventTimeSessionWindows.withGap(Time.minutes(15)))
最后，我们应用一个window function，就可以用5行代码实现乱序问题的处理：
val sessionStream : DataStream[SessionStats] = sEnv
.addSource(new FlinkKafkaConsumer09[Event](...))
.keyBy(_.userId)
.window(EventTimeSessionWindows.withGap(Time.minutes(15)))
.apply(new CountSessionStats())
上边这5行代码够优雅么？
答案是否定的，这里即使基于Event Time以及应用watermark来处理乱序，依然不够理想。考虑下面两种情况：
- 例如15分钟之后，突然来了一条之前5分钟数据怎么办？这时之前的session就不应该产生gap。
- 再比如假如一个用户一直在听音乐，gap一直没产生，那么这个用户的数据就一直无法及时产生。这个对于结果的反馈时间太长了。
对于第一种情况，Flink提供了allowedLateness来处理延迟的数据，假设我们预计有些数据最晚会延迟1小时到来，那么我们可以通过allowedLateness设置一个参数，来处理那些延迟的数据：
.allowedLateness(Time.minutes(60))
这样，当late data element到达时，我们依然可以正确的处理。
对于第二种情况，为了缩短结果的反馈时间，我们可以自定义一个early firing trigger实现每隔一段时间就触发一次计算：
.trigger(EarlyTriggeringTrigger.every(Time.minutes(10)))
例如，我们每隔10分钟就触发一次窗口计算。考虑一个简单的例子，假如一个用户session持续了1个小时，那么通过这种触发器，我们就可以每10分钟便得到一个结果，之后的结果不断更新之前的结果，最终趋于正确。后边的结果相当于对前边的结果的刷新。
尽管代码相当简单，但是其背后却是Flink内援原理的支撑，例如低延迟、高吞吐、有状态的处理、简单的tasks等。
EarlyTriggeringTrigger的实现
class EarlyTriggeringTrigger(interval: Long) extends Trigger[Object, TimeWindow] {
//通过reduce函数维护一个Long类型的数据，此数据代表即将触发的时间戳
private type JavaLong = java.lang.Long
//这里取2个注册时间的最小值，因为首先注册的是窗口的maxTimestamp，也是最后一个要触发的时间
private val min: ReduceFunction[JavaLong] = new ReduceFunction[JavaLong] {
override def reduce(value1: JavaLong, value2: JavaLong): JavaLong = Math.min(value1, value2)
}
private val serializer: TypeSerializer[JavaLong] = LongSerializer.INSTANCE.asInstanceOf[TypeSerializer[JavaLong]]
private val stateDesc = new ReducingStateDescriptor[JavaLong]("fire-time", min, serializer)
//每个元素都会运行此方法
override def onElement(element: Object,
timestamp: Long,
window: TimeWindow,
ctx: TriggerContext): TriggerResult =
//如果当前的watermark超过窗口的结束时间，则清除定时器内容，触发窗口计算
if (window.maxTimestamp <= ctx.getCurrentWatermark) {
clearTimerForState(ctx)
TriggerResult.FIRE
}
else {
//否则将窗口的结束时间注册给EventTime定时器
ctx.registerEventTimeTimer(window.maxTimestamp)
//获取当前状态中的时间戳
val fireTimestamp = ctx.getPartitionedState(stateDesc)
//如果第一次执行，则将元素的timestamp进行floor操作，取整后加上传入的实例变量interval，得到下一次触发时间并注册，添加到状态中
if (fireTimestamp.get == null) {
val start = timestamp - (timestamp % interval)
val nextFireTimestamp = start + interval
ctx.registerEventTimeTimer(nextFireTimestamp)
fireTimestamp.add(nextFireTimestamp)
}
//此时继续等待
TriggerResult.CONTINUE
}
//这里不基于processing time，因此永远不会基于processing time 触发
override def onProcessingTime(time: Long,
window: TimeWindow,
ctx: TriggerContext): TriggerResult = TriggerResult.CONTINUE
//之前注册的Event Time Timer定时器，当watermark超过注册的时间时，就会执行onEventTime方法
override def onEventTime(time: Long,
window: TimeWindow,
ctx: TriggerContext): TriggerResult = {
//如果注册的时间等于maxTimestamp时间，清空状态，并触发计算
if (time == window.maxTimestamp()) {
clearTimerForState(ctx)
TriggerResult.FIRE
} else {
//否则，获取状态中的值（maxTimestamp和nextFireTimestamp的最小值）
val fireTimestamp = ctx.getPartitionedState(stateDesc)
//如果状态中的值等于注册的时间，则删除此定时器时间戳，并注册下一个interval的时间，触发计算
//这里，前提条件是watermark超过了定时器中注册的时间，就会执行此方法，理论上状态中的fire time一定是等于注册的时间的
if (fireTimestamp.get == time) {
fireTimestamp.clear()
fireTimestamp.add(time + interval)
ctx.registerEventTimeTimer(time + interval)
TriggerResult.FIRE
} else {
//否则继续等待
TriggerResult.CONTINUE
}
}
}
//上下文中获取状态中的值，并从定时器中清除这个值
private def clearTimerForState(ctx: TriggerContext): Unit = {
val timestamp = ctx.getPartitionedState(stateDesc).get()
if (timestamp != null) {
ctx.deleteEventTimeTimer(timestamp)
}
}
//用于session window的merge，判断是否可以merge
override def canMerge: Boolean = true
override def onMerge(window: TimeWindow,
ctx: OnMergeContext): TriggerResult = {
ctx.mergePartitionedState(stateDesc)
val nextFireTimestamp = ctx.getPartitionedState(stateDesc).get()
if (nextFireTimestamp != null) {
ctx.registerEventTimeTimer(nextFireTimestamp)
}
TriggerResult.CONTINUE
}
//删除定时器中已经触发的时间戳，并调用Trigger的clear方法
override def clear(window: TimeWindow,
ctx: TriggerContext): Unit = {
ctx.deleteEventTimeTimer(window.maxTimestamp())
val fireTimestamp = ctx.getPartitionedState(stateDesc)
val timestamp = fireTimestamp.get
if (timestamp != null) {
ctx.deleteEventTimeTimer(timestamp)
fireTimestamp.clear()
}
}
override def toString: String = s"EarlyTriggeringTrigger($interval)"
}
//类中的every方法，传入interval，作为参数传入此类的构造器，时间转换为毫秒
object EarlyTriggeringTrigger {
def every(interval: Time) = new EarlyTriggeringTrigger(interval.toMilliseconds)
}
