参考资料:
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 这种也可以，但是需要一定的条件，所以这个在后续在另讲。
网友评论