Flink 中极其重要的 Time 与 Window 详细解析 (深度好文，建议收藏)

1.开启发送端口，端口号为9999nc -lk 9999
2.发送内容（key 代表不同的路口，value 代表每次通过的车辆）一次发送一行，发送的时间间隔代表汽车经过的时间间隔9,39,29,74,92,61,52,35,75,4

object Window {  def main(args: Array[String]): Unit = {    //TODO time-window    //1.创建运行环境    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //2.定义数据流来源    val text = env.socketTextStream("localhost", 9999)
    //3.转换数据格式，text->CarWc    case class CarWc(sensorId: Int, carCnt: Int)    val ds1: DataStream[CarWc] = text.map {      line => {        val tokens = line.split(",")        CarWc(tokens(0).trim.toInt, tokens(1).trim.toInt)      }    }
    //4.执行统计操作，每个sensorId一个tumbling窗口，窗口的大小为5秒    //也就是说，每5秒钟统计一次，在这过去的5秒钟内，各个路口通过红绿灯汽车的数量。    val ds2: DataStream[CarWc] = ds1      .keyBy("sensorId")      .timeWindow(Time.seconds(5))      .sum("carCnt")
    //5.显示统计结果    ds2.print()
    //6.触发流计算    env.execute(this.getClass.getName)
  }}

//1.创建运行环境val env = StreamExecutionEnvironment.getExecutionEnvironment
//2.定义数据流来源val text = env.socketTextStream("localhost", 9999)
//3.转换数据格式，text->CarWccase class CarWc(sensorId: Int, carCnt: Int)
val ds1: DataStream[CarWc] = text.map {  line => {    val tokens = line.split(",")    CarWc(tokens(0).trim.toInt, tokens(1).trim.toInt)  }}//4.执行统计操作，每个sensorId一个sliding窗口，窗口时间10秒,滑动时间5秒//也就是说，每5秒钟统计一次，在这过去的10秒钟内，各个路口通过红绿灯汽车的数量。val ds2: DataStream[CarWc] = ds1  .keyBy("sensorId")  .timeWindow(Time.seconds(10), Time.seconds(5))  .sum("carCnt")
//5.显示统计结果ds2.print()
//6.触发流计算env.execute(this.getClass.getName)

//1.创建运行环境val env = StreamExecutionEnvironment.getExecutionEnvironment
//2.定义数据流来源val text = env.socketTextStream("localhost", 9999)
//3.转换数据格式，text->CarWccase class CarWc(sensorId: Int, carCnt: Int)
val ds1: DataStream[CarWc] = text.map {  (f) => {    val tokens = f.split(",")    CarWc(tokens(0).trim.toInt, tokens(1).trim.toInt)  }}//4.执行统计操作，每个sensorId一个tumbling窗口，窗口的大小为5//按照key进行收集，对应的key出现的次数达到5次作为一个结果val ds2: DataStream[CarWc] = ds1  .keyBy("sensorId")  .countWindow(5)  .sum("carCnt")
//5.显示统计结果ds2.print()
//6.触发流计算env.execute(this.getClass.getName)

//1.创建运行环境val env = StreamExecutionEnvironment.getExecutionEnvironment
//2.定义数据流来源val text = env.socketTextStream("localhost", 9999)
//3.转换数据格式，text->CarWccase class CarWc(sensorId: Int, carCnt: Int)
val ds1: DataStream[CarWc] = text.map {  (f) => {    val tokens = f.split(",")    CarWc(tokens(0).trim.toInt, tokens(1).trim.toInt)  }}//4.执行统计操作，每个sensorId一个sliding窗口，窗口大小3条数据,窗口滑动为3条数据//也就是说，每个路口分别统计，收到关于它的3条消息时统计在最近5条消息中，各自路口通过的汽车数量val ds2: DataStream[CarWc] = ds1  .keyBy("sensorId")  .countWindow(5, 3)  .sum("carCnt")
//5.显示统计结果ds2.print()
//6.触发流计算env.execute(this.getClass.getName)

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironmentimport org.apache.flink.api.scala._import org.apache.flink.streaming.api.windowing.time.Time
object StreamWindowReduce {  def main(args: Array[String]): Unit = {    // 获取执行环境    val env = StreamExecutionEnvironment.getExecutionEnvironment
    // 创建SocketSource    val stream = env.socketTextStream("node01", 9999)
    // 对stream进行处理并按key聚合    val streamKeyBy = stream.map(item => (item, 1)).keyBy(0)
    // 引入时间窗口    val streamWindow = streamKeyBy.timeWindow(Time.seconds(5))
    // 执行聚合操作    val streamReduce = streamWindow.reduce(      (item1, item2) => (item1._1, item1._2 + item2._2)    )
    // 将聚合数据写入文件    streamReduce.print()
    // 执行程序    env.execute("TumblingWindow")  }}

    //1. 获取流处理运行环境    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //2. 构建socket流数据源，并指定IP地址和端口号    val textDataStream = env.socketTextStream("node01", 9999).flatMap(_.split(" "))
    //3. 对接收到的数据转换成单词元组    val wordDataStream = textDataStream.map(_->1)
    //4. 使用 keyBy 进行分流（分组）    val groupedDataStream: KeyedStream[(String, Int), String] = wordDataStream.keyBy(_._1)
    //5. 使用 timeWinodw 指定窗口的长度（每3秒计算一次）    val windowDataStream: WindowedStream[(String, Int), String, TimeWindow] = groupedDataStream.timeWindow(Time.seconds(3))
    //6. 实现一个WindowFunction匿名内部类    val reduceDatStream: DataStream[(String, Int)] = windowDataStream.apply(new RichWindowFunction[(String, Int), (String, Int), String, TimeWindow] {      //在apply方法中实现数据的聚合      override def apply(key: String, window: TimeWindow, input: Iterable[(String, Int)], out: Collector[(String, Int)]): Unit = {        println("hello world")        val tuple = input.reduce((t1, t2) => {          (t1._1, t1._2 + t2._2)        })        //将要返回的数据收集起来，发送回去        out.collect(tuple)      }    })    reduceDatStream.print()    env.execute()

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironmentimport org.apache.flink.api.scala._import org.apache.flink.streaming.api.windowing.time.Time
object StreamWindowFold {  def main(args: Array[String]): Unit = {    // 获取执行环境    val env = StreamExecutionEnvironment.getExecutionEnvironment
    // 创建SocketSource    val stream = env.socketTextStream("node01", 9999,'\n',3)
    // 对stream进行处理并按key聚合    val streamKeyBy = stream.map(item => (item, 1)).keyBy(0)
    // 引入滚动窗口    val streamWindow = streamKeyBy.timeWindow(Time.seconds(5))
    // 执行fold操作    val streamFold = streamWindow.fold(100){      (begin, item) =>        begin + item._2    }
    // 将聚合数据写入文件    streamFold.print()
    // 执行程序    env.execute("TumblingWindow")  }}

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironmentimport org.apache.flink.streaming.api.windowing.time.Timeimport org.apache.flink.api.scala._
object StreamWindowAggregation {  def main(args: Array[String]): Unit = {    // 获取执行环境    val env = StreamExecutionEnvironment.getExecutionEnvironment
    // 创建SocketSource    val stream = env.socketTextStream("node01", 9999)
    // 对stream进行处理并按key聚合    val streamKeyBy = stream.map(item => (item.split(" ")(0), item.split(" ")(1))).keyBy(0)
    // 引入滚动窗口    val streamWindow = streamKeyBy.timeWindow(Time.seconds(5))
    // 执行聚合操作    val streamMax = streamWindow.max(1)
    // 将聚合数据写入文件    streamMax.print()
    // 执行程序    env.execute("TumblingWindow")  }}

val env = StreamExecutionEnvironment.getExecutionEnvironment
// 从调用时刻开始给env创建的每一个stream追加时间特征env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

def allowedLateness(lateness: Time): WindowedStream[T, K, W] = {  javaStream.allowedLateness(lateness)  this}

def sideOutputLateData(outputTag: OutputTag[T]): WindowedStream[T, K, W] = {  javaStream.sideOutputLateData(outputTag)  this}

vim conf/sql-client-defaults.yaml catalogs:   - name: hive_catalog     type: hive     hive-conf-dir: /disk0/soft/hive-conf/ #该目录需要包hive-site.xml文件 

CREATE TABLE hive_catalog.flink_db.kfk_fact_bill_master_12 (      master Row<reportDate String, groupID int, shopID int, shopName String, action int, orderStatus int, orderKey String, actionTime bigint, areaName String, paidAmount double, foodAmount double, startTime String, person double, orderSubType int, checkoutTime String>,  proctime as PROCTIME()  -- PROCTIME用来和Hive时态表关联  ) WITH (   'connector' = 'kafka',   'topic' = 'topic_name',   'format' = 'json',   'properties.bootstrap.servers' = 'host:9092',   'properties.group.id' = 'flinkTestGroup',   'scan.startup.mode' = 'timestamp',   'scan.startup.timestamp-millis' = '1607844694000'  ); 

CREATE VIEW IF NOT EXISTS hive_catalog.flink_db.view_fact_bill_master as  SELECT * FROM   (select t1.*, t2.group_id, t2.shop_id, t2.group_name, t2.shop_name, t2.brand_id,        ROW_NUMBER() OVER (PARTITION BY groupID, shopID, orderKey ORDER BY actionTime desc) rn      from hive_catalog.flink_db.kfk_fact_bill_master_12 t1         JOIN hive_catalog.flink_db.dim_extend_shop_info     /*+ OPTIONS('streaming-source.enable'='true',       'streaming-source.partition.include' = 'latest',       'streaming-source.monitor-interval' = '1 h',     'streaming-source.partition-order' = 'partition-name') */    FOR SYSTEM_TIME AS OF t1.proctime AS t2 --时态表      ON t1.groupID = t2.group_id and t1.shopID = t2.shop_id      where groupID in (202042)) t  where t.rn = 1