final StreamExecutionEnvironment env      = StreamExecutionEnvironment.getExecutionEnvironrnent();// 使用处理时间env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime) ; // 使用摄入时间
env.setStrearnTimeCharacteristic(TimeCharacteristic.IngestionTime);
// 使用事件时间env.setStrearnTimeCharacteristic(TimeCharacteri stic Eve~tTime);

package cn.itcast.watermark
import org.apache.flink.api.scala._import org.apache.flink.streaming.api.TimeCharacteristicimport org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarksimport org.apache.flink.streaming.api.scala.function.WindowFunctionimport org.apache.flink.streaming.api.scala.{DataStream, OutputTag, StreamExecutionEnvironment, WindowedStream}import org.apache.flink.streaming.api.watermark.Watermarkimport org.apache.flink.streaming.api.windowing.time.Timeimport org.apache.flink.streaming.api.windowing.windows.TimeWindowimport org.apache.flink.util.Collector
/*演示水印如何解决数据乱序问题 */object SlideOutputDemo {  //样例类 CarWc(信号等id,数量)  case class CarWc(id: String, num: Int, ts: Long)  def main(args: Array[String]): Unit = {    // 1 获取运行环境    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment    //方便观察数据设置并行度为1    env.setParallelism(1)    //设置基于事件时间进行计算    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)    //2 接收数据    val socketDs = env.socketTextStream("hlink161", 9999)    // 3 transformation 接收到数据之后按照逗号切分，拿到红绿灯id,数量（通过汽车数量）    val carWcDs: DataStream[CarWc] = socketDs.map(      line => {        val arr: Array[String] = line.split(",")        // println("===="+line)        //3.1 封装样例类        CarWc(arr(0), arr(1).toInt, arr(2).toLong)      }    )    // 3.1 设置水印机制,我们自己动手实现    val waterMarkDs: DataStream[CarWc] = carWcDs.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks[CarWc]{      /** The current maximum timestamp seen so far. */      private var currentMaxTimestamp = 0L      //定义最大允许的延迟时间      private var maxOutOfOrderness=0
      /** The timestamp of the last emitted watermark. */ //巨大的负数      private var lastEmittedWatermark:Long= Long.MinValue      //3.1.1计算并生成水印      override def getCurrentWatermark: Watermark = {        // this guarantees that the watermark never goes backwards.        //计算出的watermark时间        val potentialWM = currentMaxTimestamp - maxOutOfOrderness        //保证水印时间不会回退！！        if (potentialWM >= lastEmittedWatermark)          lastEmittedWatermark = potentialWM        return new Watermark(lastEmittedWatermark)      }      //3.1.2 获取到eventtime字段      override def extractTimestamp(element: CarWc, previousElementTimestamp: Long): Long = {        //获取日志的eventtime        val timestamp = element.ts        //判断比较 eventtime是不是大于最大的eventtime时间        if (timestamp > currentMaxTimestamp)          currentMaxTimestamp = timestamp        //返回日志的eventtime时间        timestamp      }    } )
    // pre-4 定义一个侧输出流    val outputTag: OutputTag[CarWc] = new OutputTag[CarWc]("delayCarWc")    // 4 设置窗口
    val windowStream: WindowedStream[CarWc, String, TimeWindow] = waterMarkDs.keyBy(_.id).timeWindow(Time.seconds(5))      .allowedLateness(Time.seconds(5))      .sideOutputLateData(outputTag)

    //5 使用apply方法执行聚合计算    val windowRes: DataStream[CarWc] = windowStream.apply(new WindowFunction[CarWc, CarWc, String, TimeWindow] {      override def apply(key: String, window: TimeWindow, input: Iterable[CarWc], out: Collector[CarWc]): Unit = {        val iter: Iterator[CarWc] = input.iterator        println("窗口开始时间 》》"+window.getStart+"== 窗口结束时间》》"+window.getEnd+"数据 【"+input.iterator.mkString(";")+"】")        val wc: CarWc = iter.reduce((t1, t2) =>          CarWc(t1.id, t1.num + t2.num, t1.ts)        )        out.collect(wc)      }    })    windowRes.print("窗口计算结果>>")    //获取到侧输出流结果    val outputDs: DataStream[CarWc] = windowRes.getSideOutput(outputTag)    outputDs.print("侧输出流数据》》")    env.execute()  }}