小白学习flink----基于处理时间的window浅析与样例

参考资料:
https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/stream/operators/windows.html

前言

学习flink，感觉很多的坑啊，现在基本学习都是靠官方文档，但是感觉flink的官方文档不是那么的友好(也可能我太菜了)实时相对于离线来说还是难度高一些的。最近在写相关样例这里除了很多的问题，现在来一一说明一下

flink中的key

flink中的数据有一个分组的概念。我们可以使用 keyBy()针对消费的数据中的某个字段进行分组。使用keyBy 后得到一个KeyedStream，这样后算的计算就会根据分组后的数据来进行计算。
当然了我们选择性的的对数据进行分组或者不分组。 keyBy的操作要在窗口定义之前完成
这里注意一下两点

• 1、 keyBy的操作要在窗口定义之前完成
• 2、对于分组和不分组的 window方法不太一样
  分组的 window方法为 .window(...)
  不分组的window方法为 .windowAll(...)

flink的窗口类型

• 1、Tumbling Windows（滚动窗口）
  滚动窗口具有具体的大小，并且不重叠
  定义方式

.window(TumblingProcessingTimeWindows.of(Time.seconds(5))) // 这里注意，我们这里统一都按照处理时间来定义窗口，事件时间的会复杂些，后续另讲

• 2、Sliding Windows （滑动窗口）
  滑动窗口有固定的窗口大小，滑动参数来指定滑动窗口的频率。如果滑动的参数小于窗口的大小的话，滑动窗口将会重叠。
  定义方式

.window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(5))) //窗口大小 10S 滑动频率5S
//剩下两个窗口后续遇到补充

• 3、Session Windows
• 4、Global Windows

flink中的window方法

window的方法的功能是 当window指定的时候，window方法会对每一个window中的元素进行计算

window方法包含三种:ReduceFunction AggregateFunction FoldFunction ProcessWindowFunction

这几种我先讲下ReduceFunction （目前在样例中只用到这一种）

ReduceFunction

这种的书写形式

input
    .keyBy(<key selector>)
    .window(<window assigner>)
    .reduce { (v1, v2) => (v1._1, v1._2 + v2._2) }

这里需要注意一点:输出的类型必须和输入的类型一致

具体示例

简单学习了下window的用法，写了一个具体示例（只贴关键代码），下面展示下并且说一下自己遇到的问题

    val stream = env
      .addSource(kafkaConf)
      .map { x =>
        val arr = x.split(",")
        val cid = arr(0)
        val cname = arr(1)
        val ename = arr(2)
        val phone = arr(3)
        val email = arr(4)
        val address = arr(5)
        People(cid, cname, ename, phone, email, address)
      }

    //滚动窗口示例(按照cid分组统计各个cid的行数)
    val value = stream
        .map(x=>(x,1))
      .keyBy(_._1.cid)
      .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
      .reduce((x,y)=>(x._1,x._2+y._2))
      .print()

遇到的问题1 :导包错误

大家在导入包的时候 导包切记不到倒错，如果自己是使用scala书写的代码，就需要带入scala的包
我在StreamExecutionEnvironment.getExecutionEnvironment 导入 StreamExecutionEnvironment时候导入的java的这个依赖，然后运行出现问题，在这里耽误了时间

遇到的问题2 :定义窗口时间的window() 和windowAll()

大家切记 使用keyBy() 分组的时候使用window() 而不使用分组的时候使用windowAll() 我刚开始看官网有点看乱了，这一块你如果写错了它也不报错，但是它不会按照分组进行统计

遇到的问题3 : 定义窗口的时间语义

这块很重要，我刚开始直接粘贴的官网的代码.window(TumblingEventTimeWindows.of(Time.seconds(5)))来定义窗口的时间，然后跑任务跑不动，原因是因为是定义的event time而不是处理的时间，event time 必须在source上指定时间戳才可以对指定的进行处理。 后续我改成了TumblingProcessingTimeWindows 这种得以成功。
TumblingEventTimeWindows 这种也可以，但是需要一定的条件，所以这个在后续在另讲。