码迷,mamicode.com
首页 > 其他好文 > 详细

Flink去重第一弹:MapState去重

时间:2020-01-12 22:11:42      阅读:466      评论:0      收藏:0      [点我收藏+]

标签:param   date   一个   sof   kafka   timer   serve   turn   过滤   

去重计算应该是数据分析业务里面常见的指标计算,例如网站一天的访问用户数、广告的点击用户数等等,离线计算是一个全量、一次性计算的过程通常可以通过distinct的方式得到去重结果,而实时计算是一种增量、长期计算过程,我们在面对不同的场景,例如数据量的大小、计算结果精准度要求等可以使用不同的方案。此篇介绍如何通过编码方式实现精确去重,以一个实际场景为例:计算每个广告每小时的点击用户数,广告点击日志包含:广告位ID、用户设备ID(idfa/imei/cookie)、点击时间。

实现步骤分析:

  1. 为了当天的数据可重现,这里选择事件时间也就是广告点击时间作为每小时的窗口期划分

  2. 数据分组使用广告位ID+点击事件所属的小时

  3. 选择processFunction来实现,一个状态用来保存数据、另外一个状态用来保存对应的数据量

  4. 计算完成之后的数据清理,按照时间进度注册定时器清理

实现

广告数据

  1. case class AdData(id:Int,devId:String,time:Long)

分组数据

  1. case class AdKey(id:Int,time:Long)

主流程

  1. val env=StreamExecutionEnvironment.getExecutionEnvironment

  2. env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

  3.  

  4. val kafkaConfig=new Properties()

  5. kafkaConfig.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092")

  6. kafkaConfig.put(ConsumerConfig.GROUP_ID_CONFIG,"test1")

  7. val consumer=new FlinkKafkaConsumer[String]("topic1",new SimpleStringSchema,kafkaConfig)

  8. val ds=env.addSource(consumer)

  9. .map(x=>{

  10. val s=x.split(",")

  11. AdData(s(0).toInt,s(1),s(2).toLong)

  12. }).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[AdData](Time.minutes(1)) {

  13. override def extractTimestamp(element: AdData): Long = element.time

  14. })

  15. .keyBy(x=>{

  16. val endTime= TimeWindow.getWindowStartWithOffset(x.time, 0,

  17. Time.hours(1).toMilliseconds) + Time.hours(1).toMilliseconds

  18. AdKey(x.id,endTime)

  19. })

指定时间时间属性,这里设置允许1min的延时,可根据实际情况调整;
时间的转换选择TimeWindow.getWindowStartWithOffset Flink在处理window中自带的方法,使用起来很方便,第一个参数 表示数据时间,第二个参数offset偏移量,默认为0,正常窗口划分都是整点方式,例如从0开始划分,这个offset就是相对于0的偏移量,第三个参数表示窗口大小,得到的结果是数据时间所属窗口的开始时间,这里加上了窗口大小,使用结束时间与广告位ID作为分组的Key。

去重逻辑
自定义Distinct1ProcessFunction 继承了KeyedProcessFunction, 方便起见使用输出类型使用Void,这里直接使用打印控制台方式查看结果,在实际中可输出到下游做一个批量的处理然后在输出;
定义两个状态:MapState,key表示devId, value表示一个随意的值只是为了标识,该状态表示一个广告位在某个小时的设备数据,如果我们使用rocksdb作为statebackend, 那么会将mapstate中key作为rocksdb中key的一部分,mapstate中value作为rocksdb中的value, rocksdb中value 大小是有上限的,这种方式可以减少rocksdb value的大小;另外一个ValueState,存储当前MapState的数据量,是由于mapstate只能通过迭代方式获得数据量大小,每次获取都需要进行迭代,这种方式可以避免每次迭代。

  1. class Distinct1ProcessFunction extends KeyedProcessFunction[AdKey, AdData, Void] {

  2. var devIdState: MapState[String, Int] = _

  3. var devIdStateDesc: MapStateDescriptor[String, Int] = _

  4.  

  5. var countState: ValueState[Long] = _

  6. var countStateDesc: ValueStateDescriptor[Long] = _

  7.  

  8. override def open(parameters: Configuration): Unit = {

  9.  

  10. devIdStateDesc = new MapStateDescriptor[String, Int]("devIdState", TypeInformation.of(classOf[String]), TypeInformation.of(classOf[Int]))

  11. devIdState = getRuntimeContext.getMapState(devIdStateDesc)

  12.  

  13. countStateDesc = new ValueStateDescriptor[Long]("countState", TypeInformation.of(classOf[Long]))

  14. countState = getRuntimeContext.getState(countStateDesc)

  15. }

  16.  

  17. override def processElement(value: AdData, ctx: KeyedProcessFunction[AdKey, AdData, Void]#Context, out: Collector[Void]): Unit = {

  18.  

  19. val currW=ctx.timerService().currentWatermark()

  20. if(ctx.getCurrentKey.time+1<=currW) {

  21. println("late data:" + value)

  22. return

  23. }

  24.  

  25. val devId = value.devId

  26. devIdState.get(devId) match {

  27. case 1 => {

  28. //表示已经存在

  29. }

  30. case _ => {

  31. //表示不存在

  32. devIdState.put(devId, 1)

  33. val c = countState.value()

  34. countState.update(c + 1)

  35. //还需要注册一个定时器

  36. ctx.timerService().registerEventTimeTimer(ctx.getCurrentKey.time + 1)

  37. }

  38. }

  39. println(countState.value())

  40. }

  41.  

  42. override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[AdKey, AdData, Void]#OnTimerContext, out: Collector[Void]): Unit = {

  43. println(timestamp + " exec clean~~~")

  44. println(countState.value())

  45. devIdState.clear()

  46. countState.clear()

  47. }

  48. }

数据清理通过注册定时器方式ctx.timerService().registerEventTimeTimer(ctx.getCurrentKey.time + 1)表示当watermark大于该小时结束时间+1就会执行清理动作,调用onTimer方法。

在处理逻辑里面加了

  1. val currW=ctx.timerService().currentWatermark()

  2. if(ctx.getCurrentKey.time+1<=currW){

  3. println("late data:" + value)

  4. return

  5. }

主要考虑可能会存在滞后的数据比较严重,会影响之前的计算结果,做了一个类似window机制里面的一个延时判断,将延时的数据过滤掉,也可以使用OutputTag 单独处理。

技术图片

Flink去重第一弹:MapState去重

标签:param   date   一个   sof   kafka   timer   serve   turn   过滤   

原文地址:https://www.cnblogs.com/pucheung/p/12184764.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!