【Flink】关于Time和Window的基本知识点 原创

601-赵同学

发表文章数:191

首页 » 大数据 » 正文


引言

【Flink】关于Time和Window的基本知识点
                    原创
Flink有四大基石:CheckpointStateTimeWindow
这篇文章主要说一下TimeWindow
 

Time

概述

【Flink】关于Time和Window的基本知识点
                    原创
Event Time —— 时间创建的时间,通常由事件中的时间戳描述。比如采集日志数据中,每一个日志都会记录自己的生成时间,Flink通过时间戳分配器访问时间时间戳。
Ingestion Time —— 数据进入Flink的时间。
Processing Time —— 每一个执行基于时间操作的算子的本地系统时间,与机器相关,默认的时间属性就是Processing Time。

如果要统计1min内的故障日志个数,Event Time更有意义,因为要根据日志的生成时间进行统计。
 

Window

TimeWindow

滚动窗口 | Tumbling Windows

滚动窗口分配程序会将每个元素分配到一个指定窗口大小的窗口中,滚动窗口大小固定且不重叠
例如,如果指定了一个大小为5分钟的滚动窗口,那么就会计算当前窗口并每隔五分钟启动一个新的窗口,如下图所示
【Flink】关于Time和Window的基本知识点
                    原创
所以,滚动窗口的特点时间对齐窗口长度固定没有重叠。一般适用于做每个时间段的聚合计算,比如BI统计等。
 

滑动窗口 | Sliding Windows

滑动窗口分配程序会将元素分配到固定长度的窗口中,与滚动窗口类似,窗口的大小由窗口大小参数来配置,另一个附加的滑动窗口参数控制滑动窗口启动的频率。因此,如果滑动窗口小于窗口大小,则滑动窗口可以重叠。在这种情况下,元素被分配给多个窗口。
例如,有一个10分钟大小的窗口和5分钟的滑动,那么每隔5分钟就会出现一个包含最近10分钟到达的事件的窗口,如下图所示
【Flink】关于Time和Window的基本知识点
                    原创

会话窗口 | Session Windows

会话窗口分配器会按session活动对元素进行分组,会话窗口不会重叠,有没有固定的开始时间和结束时间。当会话窗口在一段时间内没有接收到元素时,也即非活动间隔产生,会话窗口就会关闭。一个session窗口由一个session间隔来配置,这个session间隔定义了非活跃周期的长度,当这个非活跃周期产生,那么当前的session将关闭并且后续的元素将被分配到新的session窗口中去。
【Flink】关于Time和Window的基本知识点
                    原创
 

CountWindow

CountWindow根据窗口中相同key元素的数量来触发执行,执行时只计算元素数量达到窗口大小的key对应的结果。
注意:CountWindow的window_size指的是相同Key的元素的个数,不是输入的所有元素的总数。
 

实例

CountWindow && TimeWindow
import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.streaming.api.scala.{DataStream, KeyedStream, StreamExecutionEnvironment, WindowedStream}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.{GlobalWindow, TimeWindow}

object WordCountFromWindows {
  def main(args: Array[String]): Unit = {
    /**
     * 1.获取执行环境
     * 2.创建socketSource
     * 3.对获取到的数据进行切割和keyBy聚合操作
     * 4.引入窗口(timeWindow和CountWindow的滚动and滑动)
     * 5.对引入窗口数据进行聚合操作
     * 6.打印或写入文件
     * 7.执行环境
     */
    // 1.获取执行环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    // 2.创建socketSource
    val socketDataStream: DataStream[String] = env.socketTextStream("node01", 9999)

    // 3.对获取到的数据进行切割和keyBy聚合操作
    import org.apache.flink.api.scala._

    val splitDataStream: DataStream[String] = socketDataStream.flatMap(item => item.split(" "))
    val mapDataStream: DataStream[(String, Int)] = splitDataStream.map(ele => (ele, 1))
    val groupedDataStream: KeyedStream[(String, Int), Tuple] = mapDataStream.keyBy(0)

    // 4.引入窗口
      // 4.1 CountWindow
        // 4.1 滚动窗口
        val countWindowTumbling: WindowedStream[(String, Int), Tuple, GlobalWindow] = groupedDataStream.countWindow(5)
        // 4.2 滑动窗口
        val countWindowSliding: WindowedStream[(String, Int), Tuple, GlobalWindow] = groupedDataStream.countWindow(5, 3)
    
      // 4.2 TimeWindow
        // 4.2.1 滚动窗口
        val timeWindowTumbling: WindowedStream[(String, Int), Tuple, TimeWindow] = groupedDataStream.timeWindow(Time.seconds(5))
        // 4.2.2 滑动窗口
        val timeWindowSliding: WindowedStream[(String, Int), Tuple, TimeWindow] = groupedDataStream.timeWindow(Time.seconds(10), Time.seconds(5))

    // 5.对引入的窗口数据进行聚合操作
    val sumDataStream: DataStream[(String, Int)] = countWindowTumbling.sum(1)
    // 6.打印或写入文件
    sumDataStream.print()
    
    // 7.运行执行环境
    env.execute("WordCountFromWindows")
  }
}

 

WindowReduce
package cn.itcast.streaming

import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow

object WindowReduceDemo {
  def main(args: Array[String]): Unit = {
    /**
     * 1.创建执行环境
     * 2.获取socket数据源
     * 3.划分窗口
     * 4.reduce操作
     * 5.输出数据
     */
    // 1.创建执行环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    // 2.获取socket数据源
    val socketDataStream: DataStream[String] = env.socketTextStream("node01", 9999)

    // 3.对数据切割聚合
    val groupedDataStream: KeyedStream[(String, Int), Tuple] = socketDataStream.flatMap(item => item.split(" ")).map(ele => (ele, 1)).keyBy(0)

    // 4.引入窗口
    val windowedStream: WindowedStream[(String, Int), Tuple, TimeWindow] = groupedDataStream.timeWindow(Time.seconds(5))

    // 5.reduce操作
    val reduceDataStream: DataStream[(String, Int)] = windowedStream.reduce((x1, x2) => (x1._1, x1._2 + x2._2))

    reduceDataStream.print()

    env.execute("WindowReduceDemo")
  }
}

 

WindowApply
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector


object WindowApplyDemo {
  def main(args: Array[String]): Unit = {
    /**
     * 1)获取流处理运行环境
     * 2)构建socket流数据源,并指定IP地址和端口号
     * 3)对接收到的数据转换成单词元组
     * 4)使用 keyBy 进行分流(分组)
     * 5)使用 timeWinodw 指定窗口的长度(每3秒计算一次)
     * 6)实现一个WindowFunction匿名内部类
     *    a.apply方法中实现聚合计算
     *    b.使用Collector.collect收集数据
     * 7)打印输出
     * 8)启动执行
     */
    // 1.获取流处理运行环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    // 2.构建socket流数据源,并指定IP地址和端口号
    val socketDataStream: DataStream[String] = env.socketTextStream("node01", 9999)

    // 3.对接收到的数据转换成单词元组
    val mapDataStream: DataStream[(String, Int)] = socketDataStream.flatMap(_.split(" ")).map((_, 1))

    // 4.使用 keyBy 进行分流(分组)
    val keyedStream: KeyedStream[(String, Int), String] = mapDataStream.keyBy(_._1)

    // 5.使用 timeWinodw 指定窗口的长度(每3秒计算一次)
    val windowedStream: WindowedStream[(String, Int), String, TimeWindow] = keyedStream.timeWindow(Time.seconds(5))

    // 6.实现一个WindowFunction匿名内部类
    val reduceStream: DataStream[(String, Int)] = windowedStream.apply(new WindowFunction[(String, Int), (String, Int), String, TimeWindow] {
      override def apply(key: String, window: TimeWindow, input: Iterable[(String, Int)], out: Collector[(String, Int)]): Unit = {
        val tuple: (String, Int) = input.reduce((t1, t2) => {
          (t1._1, t1._2 + t2._2)
        })
        out.collect(tuple)
      }
    })

    // 7.输出数据
    reduceStream.print()

    //8.启动执行环境
    env.execute("WindowApplyDemo")
  }
}

未经允许不得转载:作者:601-赵同学, 转载或复制请以 超链接形式 并注明出处 拜师资源博客
原文地址:《【Flink】关于Time和Window的基本知识点 原创》 发布于2020-11-06

分享到:
赞(0) 打赏

评论 抢沙发

评论前必须登录!

  注册



长按图片转发给朋友

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

微信扫一扫打赏

Vieu3.3主题
专业打造轻量级个人企业风格博客主题!专注于前端开发,全站响应式布局自适应模板。

登录

忘记密码 ?

您也可以使用第三方帐号快捷登录

Q Q 登 录
微 博 登 录