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)
    }
  }
}