Spark+Hbase 亿级流量分析实战（ PV/UV ）

时间 2019-12-04

标签 spark+hbase spark hbase 流量分析实战栏目 Spark 繁體版

原文原文链接

做为一个百亿级的流量实时分析统计系统怎么能没有 PV / UV 这两经典的超级玛丽亚指标呢，话说五百年前它俩但是鼻祖，咳咳...，很差意思没忍住，回归正文，大猪在上一篇已经介绍了小巧高性能ETL程序设计与实现了，到如今，咱们的数据已经落地到 Hbase 上了，并且日志的时间也已经写到 Mysql 了，万事都已经具有了，接下来咱们就要撸指标了，先从两个经典的指标开始撸。mysql

程序流程

咱们先理一下整个程序的计算流程，请看大图： git

开始计算是咱们的 Driver 程序入口github
开始计算以前检查监听 Redis 有没有收到程序退出通知，若是有程序结束，不然往下执行redis
首先去查询咱们上篇文章的 ETL loghub 日志的进度的平均时间点算法
Switch 处是判断 loghub 的时间距离咱们上次计算的指标时间是否相差足够时间，通常定义为3分钟时间以后，由于 loghub 的时间会有少许的波动状况sql
不知足则 Sleep 30秒，能够本身控制Sleep范围。apache
知足则计算 上次指标计算结束时间 ~ (loghub时间 - 3分钟日志波动)bash
计算完成更新指标结果而且更新指标计算时间，而后回到第 2 点。多线程

程序实现

先从 DriverMain 入口开始撸起app

//监听redis退出消息
while (appRunning) {
      val dbClient = new DBJdbc(props.getProperty("jdbcUrl"))
      //日志offset
      val loghubTime = dbClient.query("loghub").toLocalDateTime.minusMinutes(3)
      //指标计算offset
      val indicatorTime =dbClient.query("indicator").toLocalDateTime
      //两个时间相差(分)
      val betweenTimeMinutes = Duration.between(indicatorTime, loghubTime).toMinutes

      val format = DateTimeFormatter.ofPattern("yyyyMMddHHmmssSSS")
      //相差足够时间则进行指标运行，不然睡眠
      if (betweenTimeMinutes >= 1) {
        app.run(spark, indicatorTime, loghubTime)
        //计算完成更新指标时间
        dbClient.upsert(Map("offsetName" -> "indicator"), Update(sets = Map("time" -> loghubTime.toString)), "offset")
      } else {
        //让咱们的老大哥睡会，别太累了
        TimeUnit.SECONDS.sleep(30)
      }
    }
复制代码

从注释上看，总体思路仍是比较清晰的。

接下来咱们跟着往下看run里面的方法作了什么有意思的操做

conf.set(TableInputFormat.INPUT_TABLE, Tables.LOG_TABLE)
conf.set("TableInputFormat.SCAN_ROW_START", start)
conf.set("TableInputFormat.SCAN_ROW_START", end)
val logDS = sc.newAPIHadoopRDD(
      conf,
      classOf[TableInputFormat2],
      classOf[ImmutableBytesWritable],
      classOf[Result]
    )
      .map(tp2 => HbaseUtil.resultToMap(tp2._2))
      .map(map => {
        LogCase(
          //子case类，存放多种格式的时间
          dt = DT(
            map.get("time").toLocalDateTimeStr(),
            map.get("time").toLocalDate().toString
          ),
          `type` = map.get("type"),
          aid = map.get("aid"),
          uid = map.get("uid"),
          tid = map.get("tid"),
          ip = map.get("ip")
        )
      }).toDS()

    logDS.cache()
    logDS.createTempView("log")
    //各种指标
    new PV().run()
    new UV().run()
复制代码

start 跟 end 就是上面传下来须要查询的日志时间范围

简要说明：就是把Hbase的时间范围数据转成SparkSQL中的一张log表

在 UV 跟 PV 指标计算里面就可使用这张 log 表了

咱们看看这两个经典的指标里面到底有什么乾坤：

spark.sql(
      """ |SELECT | aid, | dt.date, | COUNT(1) as pv |FROM | log |GROUP BY | aid, | dt.date """.stripMargin)
      .rdd
      .foreachPartition(rows => {
        val props = PropsUtils.properties("db")
        val dbClient = new DBJdbc(props.getProperty("jdbcUrl"))
        rows.foreach(row => {
          dbClient.upsert(
            Map(
              "time" -> row.getAs[String]("date"),
              "aid" -> row.getAs[String]("aid")
            ),
            Update(incs = Map("pv" -> row.getAs[Long]("pv").toString)),
            "common_report"
          )
        })
        dbClient.close()
      })
复制代码

哇然一看，大哥你这也写得太简单了吧

不就是一个普通的 PV 算法，再加上分区foreachPartition操做把更到的每一行聚合的结果数据upsert到咱们的common_report指标表

group by后面跟上要聚合的维度，以上是想统计每篇文章天天的PV

从这个方法咱们就能推算出common_report长什么样了，至少有time+aid这两个惟一索引字段，还有pv这个字段，默认值确定是 0

百闻不如一见，看看表的 DDL 是否是这样子:

create table common_report
(
	id bigint auto_increment primary key,
	aid bigint not null,
	pv int default 0 null,
	uv int default 0 null,
	time date not null,
	constraint common_report_aid_time_uindex unique (aid, time)
);
复制代码

果真一点都没错。

再看 dbClient.upsert 里面大概也能猜到是实现了mysql的upsert功能，大概的sql就会生成下面格式：

INSERT INTO common_report (time, aid, pv)
VALUES ('2019-03-26', '10000', 1) ON DUPLICATE KEY UPDATE pv = pv + 1;
复制代码

大猪那 UV 是怎么实现咧？一个用户在今天来过第一次以后再来就不能重复计算了噢。

大猪答：这个简单简单，可使用Redis去重嘛，可是咱们使用的都是Hbase了，还使用它作啥子咧，具体咱们看一下 UV 里面究竟是如何实现的：

val logDS = spark.table("log").as(ExpressionEncoder[LogCase])
    import spark.implicits._
    logDS
      .mapPartitions(partitionT => {
        val hbaseClient = DBHbaseHelper.getDBHbase(Tables.CACHE_TABLE)
        val md5 = (log: LogCase) => MD5Hash.getMD5AsHex(s"${log.dt.date}|${log.aid}|${log.uid}|uv".getBytes)
        partitionT
          .grouped(Consts.BATCH_MAPPARTITIONS)
          .flatMap { tList =>
            tList
              .zip(hbaseClient.incrments(tList.map(md5)))
              .map(tp2 => {
                val log = tp2._1
                log.copy(ext = EXT(tp2._2))
              })
          }
      }).createTempView("uvTable")

    spark.sql(
      """ |SELECT | aid, | dt.date, | COUNT(1) as uv |FROM | uvTable |WHERE | ext.render = 1 |GROUP BY | aid, | dt.date """.stripMargin)
      .rdd
      .foreachPartition(rows => {
        val props = PropsUtils.properties("db")
        val dbClient = new DBJdbc(props.getProperty("jdbcUrl"))
        rows.foreach(row => {
          dbClient.upsert(
            Map(
              "time" -> row.getAs[String]("date"),
              "aid" -> row.getAs[String]("aid")
            ),
            Update(incs = Map("uv" -> row.getAs[Long]("uv").toString)),
            "common_report"
          )
        })
        dbClient.close()
      })
复制代码

spark.sql 这里跟PV同样嘛，就是多了一句条件ext.render = 1，可是上面那一大堆是啥子咧？

大猪 CACHE_TABLE 是什么来的，是Hbase一张中间表，用户存用户UV标记的，建表语句以下，由于维度都是按天，因此咱们TTL设计3天就能够了，两天也能够。

create 'CACHE_FOR_TEST',{NAME => 'info',TTL => '3 DAYS',CONFIGURATION => {'SPLIT_POLICY' => 'org.apache.hadoop.hbase.regionserver.KeyPrefixRegionSplitPolicy','KeyPrefixRegionSplitPolicy.prefix_length'=>'2'},COMPRESSION=>'SNAPPY'},SPLITS => ['20', '40', '60', '80', 'a0', 'c0', 'e0']
复制代码

那还有其它的呢？

莫慌莫慌，大猪这就慢慢解释道：

val logDS = spark.table("log").as(ExpressionEncoder[LogCase])
复制代码

上面这句的意思就是就是把log表给取出来，固然也能够经过参数传递。

下面的mapPartitions挺有意思的:

partitionT
    .grouped(1000)
        .flatMap { tList =>
          tList
            .zip(hbaseClient.incrments(tList.map(md5)))
            .map(tp2 => {
              val log = tp2._1
              log.copy(ext = EXT(tp2._2))
            })
        }
复制代码

实际上面是处理每一个分区的数据，也就是转换数据，咱们每来一条数据就要去Hbase那incrment一次，返回来的结果就是 render ，用户今天来多少次就incrment 相应的次数。

那有什么用？我直接从Hbase GET取出数据，再判断有没有，若是没有这个用户就是今天第一次来，再把这个用户PUT进Hbase打一个标记，so easy。

其实当初咱们也是这么作的，后面发现业务的东西仍是放在SQL里面一块儿写比较好，容易维护，并且incrment好处多多，由于它是带事务的，能够多线程进行修改。

并且大家也发现了GET跟PUT是两次请求操做，保证不了事务的，指标几千万的数据少了那么几条，大家都不知道我当初找它们有辛苦。

大家有没有发现render = 1的时候是表明UV（恰好等于1的时候为何是UV？这里你们要慢慢地品尝一下了，其实就是实现了GET跟PUT操做），若是render = 2的时候又能够表明今天来过两次以上的用户指标，随时扩展，就问你撸这样的代码结构爽不爽？

看看 incrments 方法实现了啥子

def incrments(incs: Seq[String], family: String = "info", amount: Int = 1): Seq[Long] = {
    if (incs.isEmpty) {
      Seq[Long]()
    } else {
      require(incs.head.length == 32, "pk require 32 length")
      val convertIncs = incs map { pk => new Increment(Bytes.toBytes(pk.take(8))).addColumn(Bytes.toBytes(family), Bytes.toBytes(pk.takeRight(24)), amount) }
      val results = new Array[Object](convertIncs.length)
      table.batch(convertIncs.asJava, results)
      results.array.indices.map(
        ind =>
          Bytes.toLong(
            results(ind)
              .asInstanceOf[Result]
              .getValue(
                Bytes.toBytes(family),
                Bytes.toBytes(incs(ind).takeRight(24))
              )
          )
      )
    }
  }
复制代码

这个方法就是实现了 incrment 的批量处理，由于咱们在线上生产环境的时候测试过，批量处理比单条处理性能高了上百倍，因此这也就是为何要写在mapPartitions里面的缘由了，由于只有在这个方法里面才有批量数据转换操做，foreachPartition是批量处理操做，foreach,与map是一条一条操做不能使用，咱们在输出报表到Mysql的地方已经用到过了。

大猪不知不觉已经写了那么长的文章了

关闭计算程序只须要给redis发一条stop消息就能够啦

RedisUtil().getResource.publish("computeListenerMessage", "stop")
复制代码

不能再复制代码了，不能显得文章是靠代码撑起来的。

福利完整项目源码