Flink 之Window介绍与使用

一、窗口概述

流式计算是一种被设计用于处理无限数据集的数据处理引擎，而无限数据集是指一种不断增长的本质上无限的数据集，而Window窗口是一种切割无限数据为有限块进行处理的手段。

Window是无限数据流处理的核心，Window将一个无限的stream拆分成有限大小的”buckets”桶，我们可以在这些桶上做计算操作。

二、窗口类型

Window可以分成两类：

1、时间窗口（TimeWindow）：按照时间生成Window，根据窗口实现原理可以分成三类：

1. 滚动窗口（Tumbling Window）

   将数据依据固定的窗口长度对数据进行切片。滚动窗口分配器将每个元素分配到一个指 定窗口大小的窗口中，滚动窗口有一个固定的大小，并且不会出现重叠。

   

   适用场景：适合做BI统计等（做每个时间段的聚合计算）

env.setParallelism(1)
val fileDS = env.socketTextStream("localhost", 9999)
val wordDS = fileDS.flatMap(_.split(" "))
val word2OneDS = wordDS.map((_,1))
val wordKS = word2OneDS.keyBy(_._1)
val wordWS = wordKS.timeWindow(Time.seconds(3))
val sumDS = wordWS.sum(1)
sumDS.print("window>>>")
env.execute()
           

1. 滑动窗口（Sliding Window）

   滑动窗口是固定窗口的更广义的一种形式，滑动窗口由固定的窗口长度和滑动间隔组 成。滑动窗口分配器将元素分配到固定长度的窗口中，与滚动窗口类似，窗口的大小由 窗口大小参数来配置，另一个窗口滑动参数控制滑动窗口开始的频率。因此，滑动窗口 如果滑动参数小于窗口大小的话，窗口是可以重叠的，在这种情况下元素会被分配到多 个窗口中。

   

   适用场景：对最近一个时间段内的统计, 比如求最近1小时内每5分钟的水位变化

env.setParallelism(1)
val fileDS = env.socketTextStream("localhost", 9999)
val wordDS = fileDS.flatMap(_.split(" "))
val word2OneDS = wordDS.map((_,1))
val wordKS = word2OneDS.keyBy(_._1)
val wordWS = wordKS.timeWindow(Time.seconds(3), Time.seconds(1))
val sumDS = wordWS.sum(1)
sumDS.print("window>>>")
env.execute()
           

1. 会话窗口（Session Window）

   由一系列事件组合一个指定时间长度的timeout间隙组成，类似于web应用的session， 也就是一段时间没有接收到新数据就会生成新的窗口。

   

val env = StreamExecutionEnvironment.getExecutionEnvironment
env.setParallelism(1)
val fileDS = env.socketTextStream("localhost", 9999)
val wordDS = fileDS.flatMap(_.split(" "))
val word2OneDS = wordDS.map((_,1))
val wordKS = word2OneDS.keyBy(_._1)
val wordWS = wordKS.window(ProcessingTimeSessionWindows.withGap(Time.seconds(2)))
val sumDS = wordWS.sum(1)
sumDS.print("window>>>")
env.execute()
           

2、计数窗口（CountWindow）：按照指定的数据条数生成一个Window，与时间无关。根据窗口实现原理可以分成两类:

1. 滚动窗口

   默认的CountWindow是一个滚动窗口，只需要指定窗口大小即可，当元素数量达到窗 口大小时，就会触发窗口的执行。

val dataDS: DataStream[String] = env.socketTextStream("hadoop01", 9999)

val mapDS = dataDS.map(data=>{
    val datas = data.split(",")
    (datas(0),1)
})

val reduceDS = mapDS.keyBy(_._1)
    .countWindow(3).reduce(
    (t1, t2) => {
        (t1._1, t1._2 + t2._2)
    }
)

reduceDS.print()
           

1. 滑动窗口

   滑动窗口和滚动窗口的函数名是完全一致的，只是在传参数时需要传入两个参数，一个 是window_size，一个是sliding_size。下面代码中的sliding_size设置为了2，也就是说， 每收到两个相同key的数据就计算一次，每一次计算的window范围是3个元素。

val dataDS = env.socketTextStream("hadoop02", 9999)

val mapDS = dataDS.map(data=>{
    val datas = data.split(",")
    (datas(0),datas(2).toInt)
})

val reduceDS = mapDS.keyBy(_._1)
    .countWindow(3,2).reduce(
    (t1, t2) => {
        (t1._1, t1._2 + t2._2)
    }
)

reduceDS.print()
           

三、窗口使用API

window function 定义了要对窗口中收集数据后所做的计算操作，主要可以分为两类：

1、增量聚合函数（incremental aggregation functions）

每条数据到来就进行计算，保持一个简单的状态。典型的增量聚合函数有：

1）ReduceFunction

val dataDS: DataStream[String] =
 env.socketTextStream("hadoop02", 9999)

val mapDS = dataDS.map(data=>{
    val datas = data.split(",")
    (datas(0),datas(2).toInt)
})

val reduceDS: DataStream[(String, Int)] = mapDS.keyBy(_._1)
    .timeWindow(Time.seconds(3)).reduce(
        new ReduceFunction[(String, Int)] {
            override def reduce(t: (String, Int), t1: (String, Int)): (String, Int) = {
                (t._1, t._2 + t1._2)
            }
        }
    )
reduceDS.print()
           

2）AggregateFunction

val dataDS: DataStream[String] =
 env.socketTextStream("hadoop02", 9999)

val mapDS = dataDS.map(data=>{
    val datas = data.split(",")
    (datas(0),datas(2).toInt)
})

val aggregateDS: DataStream[(String, Int)] = mapDS.keyBy(_._1)
    .countWindow(3).aggregate(
        // TODO 此处聚合函数类似于Spark中的累加器
        new AggregateFunction[(String, Int), (Int, Int), (String, Int)] {
            override def createAccumulator(): (Int, Int) = {
                (0,0)
            }

            override def add(in: (String, Int), acc: (Int, Int)): (Int, Int) = {
                (in._2 + acc._1, acc._2 + 1)
            }

            override def getResult(acc: (Int, Int)): (String, Int) = {
                ("sensor", (acc._1 / acc._2))
            }

            override def merge(acc: (Int, Int), acc1: (Int, Int)): (Int, Int) = {
                (acc._1 + acc1._1, acc._2 + acc1._2)
            }
        }
    )

aggregateDS.print()
           

2、全窗口函数（full window functions）

先把窗口所有数据收集起来，等到计算的时候会遍历所有数据。

ProcessWindowFunction

就是一个对整个窗口中数据处理的函数。

val dataDS: DataStream[String] =
 env.socketTextStream("hadoop02", 9999)

val mapDS = dataDS.map(data=>{
    val datas = data.split(",")
    (datas(0),datas(2).toInt)
})

val processDS: DataStream[String] = mapDS.keyBy(_._1)
    .countWindow(3)
    .process(new ProcessWindowFunction[(String, Int), String, String, GlobalWindow] {
        override def process(key: String, context: Context, elements: Iterable[(String, Int)], out: Collector[String]): Unit = {
            //println(elements.mkString(","))
            out.collect(elements.mkString(","))
        }
    })
processDS.print()
           

3、其它可选API

• .trigger() —— 触发器:定义 window 什么时候关闭，触发计算并输出结果
• .evitor() —— 移除器:定义移除某些数据的逻辑
• .allowedLateness() —— 允许处理迟到的数据
• .sideOutputLateData()—— 将迟到的数据放入侧输出流
• .getSideOutput() —— 获取侧输出流