SparkStreaming项目实战，实时计算pv和uv（硬肝）

比眉伴天荒 2022-10-17 15:54 152阅读 0赞

关注我，回复"资料"，获取大数据资料

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

关于什么是pv,uv，可以参见这篇博客:https://blog.csdn.net/petermsh/article/details/78652246

# 1、项目流程 #

![f40ffbd103056fd5a4429e520f70e5fc.png][]日志数据从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")

**猜你喜欢**

[Flink状态管理与状态一致性（长文）][Flink]

[Flink实时计算topN热榜][Flink_topN]

[Hbase集群挂掉的一次惊险经历][Hbase]

[数仓建模分层理论][Link 1]

[数仓架构发展史][Link 2]

[数仓建模方法论][Link 3]

[学习建议，大数据组件重点学习这几个][Link 4]

[f40ffbd103056fd5a4429e520f70e5fc.png]: /images/20221014/6f9fc05194154b96a092be762183aacd.png
[Flink]: http://mp.weixin.qq.com/s?__biz=MzU5NzAwNDQ4NQ%3D%3D&chksm=fe5b48a6c92cc1b0e54cf60e5daa028d24bc8776f25259925926f9cf33bb051e39249d0273ca&idx=1&mid=2247487756&scene=21&sn=f11ce03880a6e2979d4613c42f7f2188#wechat_redirect
[Flink_topN]: http://mp.weixin.qq.com/s?__biz=MzU5NzAwNDQ4NQ%3D%3D&chksm=fe5b5696c92cdf8066cd82c9d3e95d57fc60791f00153320a1d0d822aa06c9b51b59b3ac1eb6&idx=1&mid=2247487292&scene=21&sn=b6c487b0854fb5dc8a2859f917f6a339#wechat_redirect
[Hbase]: http://mp.weixin.qq.com/s?__biz=MzU5NzAwNDQ4NQ%3D%3D&chksm=fe5b4933c92cc025399f8052e15092ccedb4fb108f22d792fb93cff5b2528eba5f541c5751d7&idx=1&mid=2247487641&scene=21&sn=3029aaca08a6647f8e94031404435d11#wechat_redirect
[Link 1]: http://mp.weixin.qq.com/s?__biz=MzU5NzAwNDQ4NQ%3D%3D&chksm=fe5b525cc92cdb4a6842f5f48348b35bd10d265528c3b4861c1e89b5e09b84ab986282aa4a55&idx=1&mid=2247486454&scene=21&sn=c38a6f1a4020226adb0103bad03a82d1#wechat_redirect
[Link 2]: http://mp.weixin.qq.com/s?__biz=MzU5NzAwNDQ4NQ%3D%3D&chksm=fe5b55efc92cdcf9f44710a6431c4c9ab4ccb1d10b7d72bdb17382895c0a766aa159e7dfaf52&idx=1&mid=2247486533&scene=21&sn=61c32643ed7b1b782924f2649e800a05#wechat_redirect
[Link 3]: http://mp.weixin.qq.com/s?__biz=MzU5NzAwNDQ4NQ%3D%3D&chksm=fe5b5244c92cdb522f25d5a80e54fb5578837e78e93ad9a609e5cf8ec405d7e7ab0f0076eed1&idx=1&mid=2247486446&scene=21&sn=579828fcfd32a6a73d35f219df227ad9#wechat_redirect
[Link 4]: http://mp.weixin.qq.com/s?__biz=MzU5NzAwNDQ4NQ%3D%3D&chksm=fe5b4876c92cc1604b9be29b50a542a649cf01f422c8e7279ddeb80baac69d5620f0a6f1d427&idx=1&mid=2247487964&scene=21&sn=cc2a5da694e6077a235f9b30096d0cf1#wechat_redirect