storm中DAU实时计算方案

时间:2023-01-09 16:25:27

所就职的公司是一家互联网视频公司,存在大量的实时计算需求,计算uv,pv等一些经典的实时指标统计。由于要统计当天的实时

UV,当天的uv由于要存储当天的所有的key,面临本地内存不够用的问题,异常重启后会丢失本地缓存,造成计算结果不准确的问题。;如果使用外部缓存比如redis,memcache等,在高并发时会出现效率问题。

在不断的实践中,不断改进方案,积累了如下经验:

1.使用bitMap可以节约内存。

  使用redis的bitMap,并发时候会有问题。

a .只使用本地内存

由于reidis在数据量较大时,并发会达到瓶颈,干脆绕过redis,全部使用基于本地内存的bitMap方案。这样即省内存,又避免了redis的并发达不到性能要求的问题。

b.使用备份来解决DAU异常恢复问题。

实时任务异常时,由于历史缓存丢失,造成数据不准确问题。

定时或者定量将本地缓存的b的redis持久化的一个redis,hdfs,kafka中,这些数据只起到副本备份的作用。

当worker异常重启时,先从备份中恢复历史的历史的redis中的DAU的数据。然后再开始实时计算。

c.如何备份?备份哪些数据?


解决思路:

使用redis,hdfs等来存储备份数据,

bitMap的数据序列化,保存到文件,然后上传到hdfs上,或者redis上面。

dauBolt,接收到消息时,在execute方法中先后台恢复本地缓存,恢复成功后,再进行后续的业务处理。


使用kafka来备份当天的uuidSet  :需要备份当天的所有的uuid 

副本的备份每5分钟备份一次:使用kafka作为副本备份,可以 每1分钟或者每积累2W条UUID,就将这些UUID封装成一个msg,发送到kafka的一个topic主题队列uuidSetTopic里面,谁先到达执行条件就先执行那个? msg中要包含备份时候的时间戳发送时候的当前时间。

DauBolt:用来恢复当天到目前为止的uuidSet;dau相关的指标统计等;DauBolt会接收到uuidSetTopic的数据,还有userActionTopic的数据。

处理逻辑:

1.DauBolt的计算逻辑:

  如果接收到的是uuidSetTopic的数据:

  取出数据中的时间戳,

  判断是否是当天的备份数据,如果不是,就不处理,丢弃掉。

  判断是否本地缓存已经恢复,如果恢复,就不处理,同时通知dauSet_Spout不再读取uuidSetTopic的数据了。

                                             如果没恢复, 就解析uuidSet,并逐个恢复到bitMap中。

判断恢复完成的逻辑:

  取uuidSetTopic的数据中的时间戳dt。

  dt 和当前时间的差小于1.5分钟,(uuidSetTopic 最迟是一分钟一条。1.5分钟可以保证这个时间段内只有一个因为凑不够2w条的强制发送的)并且msg的uuid个数小于2w,则说明恢复进入尾声了,基本上追上了当前时间。

连续两次差小于1.5分钟,uuid个数小于2w,也说明追上了最新的值;

  dt 和当前时间的差 小于1.5 分钟(90s), 假设当前时间差是50s,msg的uuid个数等于2w,说明目前数据较多,不到一分钟就生产了2w条数据;接着往下取数据,将时间差的阀值缩小当前的差值50s;继续该逻辑,直到时间差大于设置的值,说明追上了最新的数据,本地缓存恢复成功。

  如果当前时间和bolt的启动时间的间隔差超过15分钟(也就是允许有15分钟来恢复dau的计算的本地缓存),也认为恢复完成(不能无限期的等待它恢复啊,但是要在日志中打印出来这种情况)。

2.如果接收到的是userAction的数据:

  取出数据中的uuid

  当前的本地缓存是否已经恢复: 如果恢复,取出uuid,判断是否在本地的bitMap中,如果不在,更新DAU的值。

                                               如果没有恢复,不进行uuid是否在bitMap的判断。

  将uuid 发送到uuidSetSave_bolt.

恢复线程需要根据时间来判断是否恢复成功了。

当storm 任务启动前,先向dauBakTopic发送一条消息,UUID为空set,时间戳为当前时间。

storm中DAU实时计算方案