import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.streaming.api.scala.{DataStream, KeyedStream, StreamExecutionEnvironment}/*** CountWindow 中的滚动窗口（Tumbling Windows）* 将数据依据固定的窗口长度对数据进行切分。*/
object TimeAndWindow {def main(args: Array[String]): Unit = {val env = StreamExecutionEnvironment.getExecutionEnvironmentval stream: DataStream[String] = env.socketTextStream("localhost",11111)val streamKeyBy: KeyedStream[(String, Long), Tuple] = stream.map(item => (item,1L)).keyBy(0)//注意：CountWindow的window_size 指的是相同key的元素的个数，不是输入的所有元素的总数。val streamWindow: DataStream[(String, Long)] = streamKeyBy.countWindow(5).reduce((item1, item2)=>(item1._1,item1._2+item2._2))streamWindow.print()env.execute("TimeAndWindow")}
}

import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.streaming.api.scala.{DataStream, KeyedStream, StreamExecutionEnvironment}/*** CountWindow 中的滑动窗口（Sliding Windows）* 将数据依据固定的窗口长度对数据进行切分。*/
object TimeAndWindow {def main(args: Array[String]): Unit = {val env = StreamExecutionEnvironment.getExecutionEnvironmentval stream: DataStream[String] = env.socketTextStream("localhost",11111)val streamKeyBy: KeyedStream[(String, Long), Tuple] = stream.map(item => (item,1L)).keyBy(0)//注意：CountWindow的window_size 指的是相同key的元素的个数，不是输入的所有元素的总数。//满足步长，就执行一次，按第一个参数的长度val streamWindow: DataStream[(String, Long)] = streamKeyBy.countWindow(5,2).reduce((item1, item2)=>(item1._1,item1._2+item2._2))streamWindow.print()env.execute("TimeAndWindow")}
}

import org.apache.flink.api.java.tuple.Tupleimport org.apache.flink.streaming.api.TimeCharacteristicimport org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractorimport org.apache.flink.streaming.api.scala._import org.apache.flink.streaming.api.windowing.assigners.{EventTimeSessionWindows, SlidingEventTimeWindows, TumblingEventTimeWindows}import org.apache.flink.streaming.api.windowing.time.Timeimport org.apache.flink.streaming.api.windowing.windows.TimeWindow

/**  * TimeWindow  */object EventTimeAndWindow {  def main(args: Array[String]): Unit = {    val env = StreamExecutionEnvironment.getExecutionEnvironment    //开启watermark    //从调用时刻开始给env创建的每一个stream追加时间特征。    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val stream: KeyedStream[(String, Long), Tuple] = env.socketTextStream("192.168.218.130", 1111).assignTimestampsAndWatermarks(      new BoundedOutOfOrdernessTimestampExtractor[String](Time.milliseconds(3000)) {        override def extractTimestamp(element: String): Long = {          // event word  eventTime是日志生成时间，我们从日志中解析EventTime          val eventTime = element.split(" ")(0).toLong          println(eventTime)          eventTime        }      }    ).map(item => (item.split(" ")(1),1L)).keyBy(0)    //加上滚动窗口，窗口大小是5s，调用window的api//    val streamWindow: WindowedStream[(String, Long), Tuple, TimeWindow] = stream.window(TumblingEventTimeWindows.of(Time.seconds(5)))    //滑动窗口//    val streamWindow: WindowedStream[(String, Long), Tuple, TimeWindow] = stream.window(SlidingEventTimeWindows.of(Time.seconds(10),Time.seconds(5)))    //会话窗口    val streamWindow: WindowedStream[(String, Long), Tuple, TimeWindow] = stream.window(EventTimeSessionWindows.withGap(Time.seconds(5)))    val streamReduce = streamWindow.reduce((item1,item2)=>(item1._1,item1._2+item2._2))    streamReduce.print()

    env.execute("EventTimeAndWindow")  }}