• 注册
  • 后端开发博客 后端开发博客 关注:0 内容:702

    SparkStreaming项目实战,实时计算pv和uv(硬肝)

  • 查看作者
  • 打赏作者
  • 当前位置: 职业司 > 后端开发 > 后端开发博客 > 正文
    • 后端开发博客
    • 本文首发于:Java大数据与数据仓库,SparkStreaming项目实战,实时计算pv和uv(硬肝)

      最近有个需求,实时统计pv,uv,结果按照date,hour,pv,uv来展示,按天统计,第二天重新统计,当然了实际还需要按照类型字段分类统计pv,uv,比如按照date,hour,pv,uv,type来展示。这里介绍最基本的pv,uv的展示。

      id uv pv date hour
      1 155599 306053 2018-07-27 18

      关于什么是pv,uv,可以参见这篇博客:blog.csdn.net/petermsh/ar…

      1、项目流程

      SparkStreaming项目实战,实时计算pv和uv(硬肝)
      日志数据从flume采集过来,落到hdfs供其它离线业务使用,也会sink到kafka,sparkStreaming从kafka拉数据过来,计算pv,uv,uv是用的redis的set集合去重,最后把结果写入mysql数据库,供前端展示使用。

      2、具体过程

      1)pv的计算

      拉取数据有两种方式,基于received和direct方式,这里用direct直拉的方式,用的mapWithState算子保存状态,这个算子与updateStateByKey一样,并且性能更好。当然了实际中数据过来需要经过清洗,过滤,才能使用。

      定义一个状态函数

      // 实时流量状态更新函数
      val mapFunction = (datehour:String, pv:Option[Long], state:State[Long]) => {
      val accuSum = pv.getOrElse(0L) + state.getOption().getOrElse(0L)
      val output = (datehour,accuSum)
      state.update(accuSum)
      output
      }
      复制代码

      这样就很容易的把pv计算出来了。

      2)uv的计算

      uv是要全天去重的,每次进来一个batch的数据,如果用原生的reduceByKey或者groupByKey对配置要求太高,在配置较低情况下,我们申请了一个93G的redis用来去重,原理是每进来一条数据,将date作为key,guid加入set集合,20秒刷新一次,也就是将set集合的尺寸取出来,更新一下数据库即可。

      helper_data.foreachRDD(rdd => {
      rdd.foreachPartition(eachPartition => {
      // 获取redis连接
      val jedis = getJedis
      eachPartition.foreach(x => {
      // 省略若干...
      jedis.sadd(key,x._2)
      // 设置存储每天的数据的set过期时间,防止超过redis容量,这样每天的set集合,定期会被自动删除
      jedis.expire(key,ConfigFactory.rediskeyexists)
      })
      // 关闭连接
      closeJedis(jedis)
      })
      })
      复制代码

      3)结果保存到数据库

      结果保存到mysql,数据库,10秒刷新一次数据库,前端展示刷新一次,就会重新查询一次数据库,做到实时统计展示pv,uv的目的。

      /**
      * 插入数据
      * @param data (addTab(datehour)+helperversion)
      * @param tbName
      * @param colNames
      */
      def insertHelper(data: DStream[(String, Long)], tbName: String, colNames: String*): Unit = {
      data.foreachRDD(rdd => {
      val tmp_rdd = rdd.map(x => x._1.substring(11, 13).toInt)
      if (!rdd.isEmpty()) {
      val hour_now = tmp_rdd.max() // 获取当前结果中最大的时间,在数据恢复中可以起作用
      rdd.foreachPartition(eachPartition => {
      try {
      val jedis = getJedis
      val conn = MysqlPoolUtil.getConnection()
      conn.setAutoCommit(false)
      val stmt = conn.createStatement()
      eachPartition.foreach(x => {
      // val sql = ....
      // 省略若干
      stmt.addBatch(sql)
      })
      closeJedis(jedis)
      stmt.executeBatch() // 批量执行sql语句
      conn.commit()
      conn.close()
      } catch {
      case e: Exception => {
      logger.error(e)
      logger2.error(HelperHandle.getClass.getSimpleName + e)
      }
      }
      })
      }
      })
      }
      // 计算当前时间距离次日零点的时长(毫秒)
      def resetTime = {
      val now = new Date()
      val todayEnd = Calendar.getInstance
      todayEnd.set(Calendar.HOUR_OF_DAY, 23) // Calendar.HOUR 12小时制
      todayEnd.set(Calendar.MINUTE, 59)
      todayEnd.set(Calendar.SECOND, 59)
      todayEnd.set(Calendar.MILLISECOND, 999)
      todayEnd.getTimeInMillis - now.getTime
      }
      复制代码

      4)数据容错

      流处理消费kafka都会考虑到数据丢失问题,一般可以保存到任何存储系统,包括mysql,hdfs,hbase,redis,zookeeper等到。这里用SparkStreaming自带的checkpoint机制来实现应用重启时数据恢复。

      checkpoint

      这里采用的是checkpoint机制,在重启或者失败后重启可以直接读取上次没有完成的任务,从kafka对应offset读取数据。

      // 初始化配置文件
      ConfigFactory.initConfig()
      val conf = new SparkConf().setAppName(ConfigFactory.sparkstreamname)
      conf.set("spark.streaming.stopGracefullyOnShutdown","true")
      conf.set("spark.streaming.kafka.maxRatePerPartition",consumeRate)
      conf.set("spark.default.parallelism","24")
      val sc = new SparkContext(conf)
      while (true){
      val ssc = StreamingContext.getOrCreate(ConfigFactory.checkpointdir + DateUtil.getDay(0),getStreamingContext _ )
      ssc.start()
      ssc.awaitTerminationOrTimeout(resetTime)
      ssc.stop(false,true)
      }
      复制代码

      checkpoint是每天一个目录,在第二天凌晨定时销毁StreamingContext对象,重新统计计算pv,uv。

      注意:ssc.stop(false,true)表示优雅地销毁StreamingContext对象,不能销毁SparkContext对象,ssc.stop(true,true)会停掉SparkContext对象,程序就直接停了。

      应用迁移或者程序升级

      在这个过程中,我们把应用升级了一下,比如说某个功能写的不够完善,或者有逻辑错误,这时候都是需要修改代码,重新打jar包的,这时候如果把程序停了,新的应用还是会读取老的checkpoint,可能会有两个问题:

      1. 执行的还是上一次的程序,因为checkpoint里面也有序列化的代码;
      2. 直接执行失败,反序列化失败;

      其实有时候,修改代码后不用删除checkpoint也是可以直接生效,经过很多测试,我发现如果对数据的过滤操作导致数据过滤逻辑改变,还有状态操作保存修改,也会导致重启失败,只有删除checkpoint才行,可是实际中一旦删除checkpoint,就会导致上一次未完成的任务和消费kafka的offset丢失,直接导致数据丢失,这种情况下我一般这么做。

      这种情况一般是在另外一个集群,或者把checkpoint目录修改下,我们是代码与配置文件分离,所以修改配置文件checkpoint的位置还是很方便的。然后两个程序一起跑,除了checkpoint目录不一样,会重新建,都插入同一个数据库,跑一段时间后,把旧的程序停掉就好。以前看官网这么说,只能记住不能清楚明了,只有自己做时才会想一下办法去保证数据准确。

      5)保存offset到mysql

      如果保存offset到mysql,就可以将pv, uv和offset作为一条语句保存到mysql,从而可以保证exactly-once语义。

      var messages: InputDStream[ConsumerRecord[String, String]] = null
      if (tpMap.nonEmpty) {
      messages = KafkaUtils.createDirectStream[String, String](
      ssc
      , LocationStrategies.PreferConsistent
      , ConsumerStrategies.Subscribe[String, String](topics, kafkaParams, tpMap.toMap)
      )
      } else {
      messages = KafkaUtils.createDirectStream[String, String](
      ssc
      , LocationStrategies.PreferConsistent
      , ConsumerStrategies.Subscribe[String, String](topics, kafkaParams)
      )
      }
      messages.foreachRDD(rdd => {
      ....
      })
      复制代码

      从mysql读取offset并且解析:

        /**
      * 从mysql查询offset
      *
      * @param tbName
      * @return
      */
      def getLastOffsets(tbName: String): mutable.HashMap[TopicPartition, Long] = {
      val sql = s"select offset from ${tbName} where id = (select max(id) from ${tbName})"
      val conn = MysqlPool.getConnection(config)
      val psts = conn.prepareStatement(sql)
      val res = psts.executeQuery()
      var tpMap: mutable.HashMap[TopicPartition, Long] = mutable.HashMap[TopicPartition, Long]()
      while (res.next()) {
      val o = res.getString(1)
      val jSONArray = JSON.parseArray(o)
      jSONArray.toArray().foreach(offset => {
      val json = JSON.parseObject(offset.toString)
      val topicAndPartition = new TopicPartition(json.getString("topic"), json.getInteger("partition"))
      tpMap.put(topicAndPartition, json.getLong("untilOffset"))
      })
      }
      MysqlPool.closeCon(res, psts, conn)
      tpMap
      }
      复制代码

      6)日志

      日志用的log4j2,本地保存一份,ERROR级别的日志会通过邮件发送到手机,如果错误太多也会被邮件轰炸,需要注意。

      val logger = LogManager.getLogger(HelperHandle.getClass.getSimpleName)
      // 邮件level=error日志
      val logger2 = LogManager.getLogger("email")
      复制代码

      请登录之后再进行评论

      登录

      手机阅读天地(APP)

      • 微信公众号
      • 微信小程序
      • 安卓APP
      手机浏览,惊喜多多
      匿名树洞,说我想说!
      问答悬赏,VIP可见!
      密码可见,回复可见!
      即时聊天、群聊互动!
      宠物孵化,赠送礼物!
      动态像框,专属头衔!
      挑战/抽奖,金币送不停!
      赶紧体会下,不会让你失望!
    • 实时动态
    • 签到
    • 做任务
    • 发表内容
    • 偏好设置
    • 到底部
    • 帖子间隔 侧栏位置:
    • 还没有账号?点这里立即注册