两种方法指定：

• collectWithTimestamp() 生成时间戳 第一个参数指数据，第二个参数是时间戳
• emitWatermark 创建watermark
• 在addSource中通过实现SourceFunction接口,复写run方法
• source code

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.sink.PrintSinkFunction
import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.watermark.Watermark
import org.apache.flink.streaming.api.scala._
object watermark {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.createLocalEnvironment()
    //设置时间特性为event time
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    val input = List(("a", 1L, 1), ("b", 1L, 1), ("b", 3L, 1))
    val source: DataStream[(String,Long,Int)] = env.addSource(
      new SourceFunction[(String,Long,Int)]() {
        override def run(sourceContext: SourceFunction.SourceContext[(String, Long, Int)]): Unit = {

          input.foreach(value=>{
            //指定时间戳 tuple第二个元素作为time stamp
            sourceContext.collectWithTimestamp(value,value._2)
            //生成watermarks watermark 为time stamp -1
            sourceContext.emitWatermark(new Watermark(value._2-1))

          })
          sourceContext.emitWatermark(new Watermark(Long.MaxValue))
        }

        override def cancel(): Unit = {}
      })

    source.print()
    env.execute("test")
  }
}

• 如果定义了外部数据源connector，就不能再实现source function接口了
• watermark分为两种类型：Periodic Watermark 和 Punctuated Watermark
• 前者根据时间周期生成，后者根据接入数据的数量生成

两种Periodic Watermark Assigner，一种为升序模式，会将数据中的Timestamp根据指定字段提取，并用当前的Timestamp作为最新的Watermark，这种Timestamp Assigner比较适合于事件按顺序生成，没有乱序事件的情况；另外一种是通过设定固定的时间间隔来指定Watermark落后于Timestamp的区间长度，也就是最长容忍迟到多长时间内的数据到达系统。

Flink提供了两个预定义实现类来生成水印

• AscendingTimestampExtractor 适用于时间戳递增的情况
• BoundedOutOfOrdernessTimestampExtractor 适用于乱序但最大延迟已知的情况

按照数据的字段来分配timestamp

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.sink.SinkFunction
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time

object ascendingTimestamps {
  def main(Args:Array[String]): Unit = {
    val env = StreamExecutionEnvironment.createLocalEnvironment()
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    val input = env.fromCollection(List(("a", 1L, 1), ("b", 1L, 1), ("b", 3L, 1)))
    val inputWithTimestampAndWatermark = input.assignAscendingTimestamps(t =>
              t._3)
    val result = inputWithTimestampAndWatermark
      .keyBy(0) //key by first element
      .timeWindow(Time.seconds(10))
      .sum("_2") //sum by second element
    result.print()
    env.execute()
  }
}

• AssignerWithPeriodicWatermarks 周期性分配timestamp 和 watermark
• ExecutionConfig.setAutoWatermarkInterval(…) 定义产生的watermark 事件间隔
• 定义BoundedOutOfOrdernessTimestampExtractor抽象类实现AssignerWithPeriodicWatermarks接口的extractTimestamp及getCurrentWatermark方法，同时声明抽象方法extractAscendingTimestamp供子类实现
• maxOutOfOrderness参数表示乱序数据允许到达且被处理的滞后最大时间(t-t_w,t为eventtime,t_w为上一个watermark的时间) 当进行窗口计算的时候，如果时间超过了最大时间，那么会被忽略。

• TPS很高的场景下会产生大量的Watermark在一定程度上对下游算子造成压力，在实时性要求非常高的场景才会选择间断的方式进行水印的生成。
• ExecutionConfig.setAutoWatermarkInterval(…) 定义产生的watermark 事件间隔

• 窗口抽象成独立的Operator
• 每个窗口算子包含了一下组成部分：
• ·Windows Assigner：指定窗口的类型，定义如何将数据流分配到一个或多个窗口
• ·Windows Trigger：指定窗口触发的时机，定义窗口满足什么样的条件触发计算；
• ·Evictor：用于数据剔除
• ·Lateness：标记是否处理迟到数据，当迟到数据到达窗口中是否触发计算；
• ·Output Tag：标记输出标签，然后在通过getSideOutput将窗口中的数据根据标签输出；
• ·Windows Funciton：定义窗口上数据处理的逻辑，例如对数据进行sum操作。

• Flink会根据上游数据集是否为KeyedStream类型(数据集按照key分区)，对应的Window Assigner也会不同。
• 具体分为Keyed和NonKeyed类型
• 前者使用window()方法指定window assigner ，然后根据key在不同的Task实例中进行并行计算

inputstream.keyBy(t=>t.id).window(new myWindowAssigner())

• 后者使用windowAll()指定window assigner

inputstream.windowAll(new myWindowAssigner())

时间窗口 与 数量窗口

Tumbling Windows
根据固定时间或大小进行切分，且窗口和窗口之间的元素互不重叠

DataStream<T> input = ...;

// tumbling event-time windows
input
    .keyBy(<key selector>)
    .window(TumblingEventTimeWindows.of(Time.seconds(5)))
    .<windowed transformation>(<window function>);

// tumbling processing-time windows
input
    .keyBy(<key selector>)
    .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
    .<windowed transformation>(<window function>);

// daily tumbling event-time windows offset by -8 hours.
input
    .keyBy(<key selector>)
    .window(TumblingEventTimeWindows.of(Time.days(1), Time.hours(-8)))
    .<windowed transformation>(<window function>);

Sliding Windows

• 窗口大小固定但是会重叠
• 元素可能会属于不同的窗口

DataStream<T> input = ...;

// sliding event-time windows
input
    .keyBy(<key selector>)
    .window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
    .<windowed transformation>(<window function>);

// sliding processing-time windows
input
    .keyBy(<key selector>)
    .window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(5)))
    .<windowed transformation>(<window function>);

// sliding processing-time windows offset by -8 hours
input
    .keyBy(<key selector>)
    .window(SlidingProcessingTimeWindows.of(Time.hours(12), Time.hours(1), Time.hours(-8)))
    .<windowed transformation>(<window function>);

第一第二个例子of(param1,param2)

第一个参数是窗口大小

第二个参数是窗口每次滑动的大小

第三个例子有第三个参数，表示offset

举个例子，假如是windows.of(1Hour,30min,15min)

1:15:00.000 - 2:14:59.999, 1:45:00.000 - 2:44:59.999 etc.

An important use case for offsets is to adjust windows to timezones other than UTC-0. For example, in China you would have to specify an offset of Time.hours(-8).

Session Windows

• 按照事件来分组
• 没有固定的开始和结束时间
• 当一段时间没有接收到时间的时候便会自动关闭

DataStream<T> input = ...;

// event-time session windows with static gap
input
    .keyBy(<key selector>)
    .window(EventTimeSessionWindows.withGap(Time.minutes(10)))
    .<windowed transformation>(<window function>);
    
// event-time session windows with dynamic gap
input
    .keyBy(<key selector>)
    .window(EventTimeSessionWindows.withDynamicGap((element) -> {
        // determine and return session gap
    }))
    .<windowed transformation>(<window function>);

// processing-time session windows with static gap
input
    .keyBy(<key selector>)
    .window(ProcessingTimeSessionWindows.withGap(Time.minutes(10)))
    .<windowed transformation>(<window function>);
    
// processing-time session windows with dynamic gap
input
    .keyBy(<key selector>)
    .window(ProcessingTimeSessionWindows.withDynamicGap((element) -> {
        // determine and return session gap
    }))
    .<windowed transformation>(<window function>);

内部机理：