val env = StreamExecutionEnvironment.getExecutionEnvironment
env.setParallelism(1)
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) //设置时间为事件时间，默认采用process
//采用socketTextStream输入
val socketDStream = env.socketTextStream("localhost",8888)
val mapDStream = socketDStream.map {
  item =>
    val arr = item.split(",")
    Item(arr(0), arr(1).toLong, arr(2).toDouble)
}
val waterDStream = mapDStream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[Item](Time.seconds(1)) { //1s为结束延时时间
  override def extractTimestamp(element: Item): Long = element.timestap * 1000L //抽取timestap作为时间标准
})
val resultDStream = waterDStream.keyBy(_.id)
  .timeWindow(Time.seconds(6), Time.seconds(3))
  .min(2) //当currentEverntTime - 1s(late延时) > window_end_time触发
resultDStream.print("bound min")
env.execute("executor")

item_1,1596419969,33.1  
item_1,1596419970,32.1
item_1,1596419971,31.1  //触发上一个窗口的的window_end操作，1596419971-1 = 1596419970(window_end_time)  [1596419964,1596419970) 此时输出的结果为33.1
item_1,1596419972,30.1 
item_1,1596419973,31.1 
item_1,1596419974,32.1 //触发上一个窗口的的window_end操作，1596419974-1 = 1596419973(window_end_time)  [1596419967,1596419973) 此时输出的结果为30.1
item_1,1596419977,29.1 //触发上一个窗口的的window_end操作，1596419977-1 = 1596419976(window_end_time)  [1596419970,1596419976) 此时输出的结果为30.1

@Override
public Collection<TimeWindow> assignWindows(Object element, long timestamp, WindowAssignerContext context) {
    timestamp = context.getCurrentProcessingTime();
    List<TimeWindow> windows = new ArrayList<>((int) (size / slide));
    long lastStart = TimeWindow.getWindowStartWithOffset(timestamp, offset, slide); //滑动步长为3s，当时间戳为1596419969时lastStart=1596419967
    for (long start = lastStart;
         start > timestamp - size;
         start -= slide) {
        //此时可以得到窗口64~70，67~73
         windows.add(new TimeWindow(start, start + size));
    }
    return windows;
}

public static long getWindowStartWithOffset(long timestamp, long offset, long windowSize) { //时间戳   偏移量(时区)默认为0 滑动步长
    return timestamp - (timestamp - offset + windowSize) % windowSize;
}  

object PeriodicInsertWatermarks {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    env.setParallelism(1)
    val stream = env
      .socketTextStream("hadoop102", 7777)
      .map {
        item =>
          val arr: Array[String] = item.split(",")
          Item(arr(0).trim, arr(1).trim.toLong, arr(2).trim.toDouble)
      }.assignTimestampsAndWatermarks(new MyAssigner)
      .keyBy(_.id)
      .timeWindow(Time.seconds(5)) //滚动窗口
      .min(2)

    stream.print()
    env.execute()
  }

  // `BoundedOutOfOrdernessTimestampExtractor`的底层实现
  class MyAssigner extends AssignerWithPeriodicWatermarks[Item] {
    val bound = 1000L // 最大延迟时间
    var maxTs = Long.MinValue + bound // 观察到的最大时间戳

    // 每来一条元素就要调用一次
    override def extractTimestamp(t: Item, l: Long): Long = {
      maxTs = maxTs.max(t.timestamp * 1000)
      t.timestamp * 1000
    }
    // 产生水位线的函数，默认200ms调用一次
    override def getCurrentWatermark: Watermark = {
      val water: Long = maxTs - bound
      println("当前水位值为:\t" + water)
      // 水位线 = 观察到的最大时间戳 - 最大延迟时间
      new Watermark(water)
    }
  }
}
case class Item(id:String,timestamp: Long,rate:Double)

class PunctuatedAssigner extends AssignerWithPunctuatedWatermarks[Item] {
  val bound: Long = 1 * 1000
  override def checkAndGetNextWatermark(r: Item, extractedTS: Long): Watermark = {
    //触发watermark的条件为id
    if (r.id == "item_1") {
      new Watermark(extractedTS - bound)
    } else {
      null
    }
  }
  override def extractTimestamp(r: Item, previousTS: Long): Long = {
    r.timestamp * 1000
  }
}

ProcessFunction
KeyedProcessFunction
CoProcessFunction
ProcessJoinFunction
BroadcastProcessFunction
KeyedBroadcastProcessFunction
ProcessWindowFunction
ProcessAllWindowFunction

currentProcessingTime(): Long 返回当前处理时间
currentWatermark(): Long 返回当前 watermark 的时间戳
registerProcessingTimeTimer(timestamp: Long): Unit 会注册当前 key 的 processing time 的定时器。 当 processing time 到达定时时间时，触发 timer。
registerEventTimeTimer(timestamp: Long): Unit 会注册当前 key 的 event time定时器。当水位线大于等于定时器注册的时间时，触发定时器执行回调函数。
deleteProcessingTimeTimer(timestamp: Long): Unit 删除之前注册处理时间定时器。如果没有这个时间戳的定时器，则不执行。
deleteEventTimeTimer(timestamp: Long): Unit 删除之前注册的事件时间定时器，如果没有此时间戳的定时器，则不执行。
当定时器 timer 触发时，会执行回调函数 onTimer()。注意定时器 timer 只能在 keyed streams 上面使用。

object KeyProcessFunctionDemo01 {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val stream = env
      .socketTextStream("localhost", 7777)
      .map {
        item =>
          val arr: Array[String] = item.split(",")
          Item(arr(0).trim, arr(1).trim.toLong, arr(2).trim.toDouble)
      }
      .keyBy(_.id)
      .process(new MyKeyedProcess)

    stream.print()
    env.execute()
  }

}
//需求：监控item的评分，如果评分值在3s之内(processing time)连续上升，则报喜。
class MyKeyedProcess extends KeyedProcessFunction[String,Item,String]{
  //保存定时器的时间戳
  lazy val lastTime = getRuntimeContext.getState(new ValueStateDescriptor[Long]("time",Types.of[Long]))
  //保存上次的评分
  lazy val lastRate = getRuntimeContext.getState(new ValueStateDescriptor[Double]("rate",Types.of[Double]))
  override def processElement(i: Item, context: KeyedProcessFunction[String, Item, String]#Context, collector: Collector[String]): Unit = {
    val time: Long = lastTime.value()
    val rate: Double = lastRate.value()
    val currentRate: Double = i.rate
    val currentTime: Long = context.timerService().currentProcessingTime() //当前时间
    val nextTimer: Long = currentTime+3000 //
    println("定时器时间为:\t"+time)
    println("当前时间为:\t\t"+currentTime)
    if (rate != 0.0 && rate < currentRate && currentTime<time ) {
      //如果上一次评分不为0(不是第一次进入) 并且 本次评分大于上次评分  并且当前时间在定时器时间之内
      context.timerService().registerProcessingTimeTimer(nextTimer)
    }
    lastTime.update(nextTimer)
    lastRate.update(currentRate)
  }
  //定时回调
  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[String, Item, String]#OnTimerContext, out: Collector[String]): Unit = {
    out.collect("当前key:\t"+ctx.getCurrentKey+"\t报喜") //输出报喜信息
  }
}
case class Item(id:String, timestap:Long, rate:Double)

val outPutDStream:DataStream[Item] = sourceDStream.process(new MySideOutputProcess)
val lowDStream:DataStream[Item] = outPutDStream.getSideOutput(new OutputTag[Item]("low"))
val highDStream:DataStream[Item] = outPutDStream.getSideOutput(new OutputTag[Item]("high"))
lowDStream.print("low")
highDStream.print("high")

//按照item的评分30为分界线，大于30输出high的item，小于0输出low的item
class MySideOutputProcess extends ProcessFunction[Item,Item]{
  //为数据打标
  lazy val lowOutput = new OutputTag[Item]("low")
  lazy val highOutput = new OutputTag[Item]("high")
  //处理每条输入数据
  override def processElement(value: Item, ctx: ProcessFunction[Item, Item]#Context, out: Collector[Item]): Unit = {
    if (value.rate < 30.0) {
      ctx.output(lowOutput,value)
    }else{
      ctx.output(highOutput,value)
    }
  }
}