活了二十多年,没能为祖国、为人民做点什么,每思及此,伤心欲绝 !

使用scala编写flink消费kafka实时计算pv,uv

大数据 keguang 417℃ 0评论

实时统计pv、uv是再常见不过的大数据统计需求了,前面出过一篇SparkStreaming实时统计pv,uv的案例,这里用flink实时计算pv,uv。

我们需要统计不同数据类型每天的pv,uv情况,并且有如下要求.

  • 每秒钟要输出最新的统计结果
  • 程序永远跑着不会停,所以要定期清理内存里的过时数据
  • 收到的消息里的时间字段并不是按照顺序严格递增的,所以要有一定的容错机制
  • 访问uv并不一定每秒钟都会变化,重复输出对IO是巨大的浪费,所以要在uv变更时在一秒内输出结果,未变更时不输出

flink数据流上的类型和操作

DataStream是flink流处理最核心的数据结构,其它的各种流都可以直接或者间接通过DataStream来完成相互转换,一些常用的流直接的转换关系如图:

可以看出,DataStream可以与KeyedStream相互转换,KeyedStream可以转换为WindowedStream,DataStream不能直接转换为WindowedStream,WindowedStream可以直接转换为DataStream。各种流之间虽然不能相互直接转换,但是都可以通过先转换为DataStream,再转换为其它流的方法来实现。

在这个计算pv,uv的需求中就主要用到DataStream、KeyedStream以及WindowedStream这些数据结构。

这里需要用到window和watermark,使用窗口把数据按天分割,使用watermark可以通过“水位”来定期清理窗口外的迟到数据,起到清理内存的作用。

业务代码

我们的数据是json类型的,含有date,helperversion,guid这3个字段,在实时统计pv,uv这个功能中,其它字段可以直接丢掉,当然了在离线数据仓库中,所有有含义的业务字段都是要保留到hive当中的。
其它相关概念就不说了,会专门介绍,这里直接上代码吧。
pom.xml文件

 

由于包含了很多其它的非flink的依赖,可以选择flink的依赖,减少下载依赖的时间。

主要代码,主要使用scala开发:

使用List集合的size保存pv,使用Set集合的size保存uv,从而达到实时统计pv,uv的目的。

存在的问题

显然,当数据量很大的时候,这个List集合和Set集合会很大,并且这里的pv是否可以不用List来存储,而是通过一个状态变量,不断做累加,对应操作就是更新状态来完成。

改进版

使用了一个计数器来存储pv的值。

我的微信公众号,专注于大数据分析与挖掘,感兴趣可以关注,看一看,瞧一瞧!

参考资料
https://flink.sojb.cn/dev/event_time.html
http://wuchong.me/blog/2016/05/20/flink-internals-streams-and-operations-on-streams/
https://segmentfault.com/a/1190000006235690

转载请注明:柯广的博客 » 使用scala编写flink消费kafka实时计算pv,uv

喜欢 (0)or分享 (0)
头像
发表我的评论
取消评论

表情

Hi,您需要填写昵称和邮箱!

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址