關(guān)注我，回復(fù)"資料"，獲取大數(shù)據(jù)資料

最近有個需求，實(shí)時統(tǒng)計(jì)pv,uv，結(jié)果按照date,hour,pv,uv來展示，按天統(tǒng)計(jì)，第二天重新統(tǒng)計(jì)，當(dāng)然了實(shí)際還需要按照類型字段分類統(tǒng)計(jì)pv,uv,比如按照date,hour,pv,uv,type來展示。這里介紹最基本的pv,uv的展示。

關(guān)于什么是pv,uv，可以參見這篇博客:https://blog.csdn.net/petermsh/article/details/78652246

1、項(xiàng)目流程

日志數(shù)據(jù)從flume采集過來，落到hdfs供其它離線業(yè)務(wù)使用，也會sink到kafka，sparkStreaming從kafka拉數(shù)據(jù)過來，計(jì)算pv,uv，uv是用的redis的set集合去重，最后把結(jié)果寫入mysql數(shù)據(jù)庫，供前端展示使用。

2、具體過程

1）pv的計(jì)算

拉取數(shù)據(jù)有兩種方式，基于received和direct方式，這里用direct直拉的方式，用的mapWithState算子保存狀態(tài)，這個算子與updateStateByKey一樣，并且性能更好。當(dāng)然了實(shí)際中數(shù)據(jù)過來需要經(jīng)過清洗，過濾，才能使用。

定義一個狀態(tài)函數(shù)

// 實(shí)時流量狀態(tài)更新函數(shù)
  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計(jì)算出來了。

2）uv的計(jì)算

uv是要全天去重的，每次進(jìn)來一個batch的數(shù)據(jù)，如果用原生的reduceByKey或者groupByKey對配置要求太高，在配置較低情況下，我們申請了一個93G的redis用來去重，原理是每進(jìn)來一條數(shù)據(jù)，將date作為key，guid加入set集合，20秒刷新一次，也就是將set集合的尺寸取出來，更新一下數(shù)據(jù)庫即可。

helper_data.foreachRDD(rdd => {
        rdd.foreachPartition(eachPartition => {
        // 獲取redis連接
          val jedis = getJedis
          eachPartition.foreach(x => {
            // 省略若干...
            jedis.sadd(key,x._2)
            // 設(shè)置存儲每天的數(shù)據(jù)的set過期時間，防止超過redis容量，這樣每天的set集合，定期會被自動刪除
            jedis.expire(key,ConfigFactory.rediskeyexists)
          })
          // 關(guān)閉連接
          closeJedis(jedis)
        })
      })

3）結(jié)果保存到數(shù)據(jù)庫

結(jié)果保存到mysql，數(shù)據(jù)庫，10秒刷新一次數(shù)據(jù)庫，前端展示刷新一次，就會重新查詢一次數(shù)據(jù)庫，做到實(shí)時統(tǒng)計(jì)展示pv,uv的目的。

/**
 * 插入數(shù)據(jù)
    * @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() // 獲取當(dāng)前結(jié)果中最大的時間，在數(shù)據(jù)恢復(fù)中可以起作用
        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() // 批量執(zhí)行sql語句
            conn.commit()
            conn.close()
          } catch {
            case e: Exception => {
              logger.error(e)
              logger2.error(HelperHandle.getClass.getSimpleName + e)
            }
          }
        })
      }
    })
  }
  
// 計(jì)算當(dāng)前時間距離次日零點(diǎn)的時長（毫秒）
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）數(shù)據(jù)容錯

流處理消費(fèi)kafka都會考慮到數(shù)據(jù)丟失問題，一般可以保存到任何存儲系統(tǒng)，包括mysql,hdfs,hbase,redis,zookeeper等到。這里用SparkStreaming自帶的checkpoint機(jī)制來實(shí)現(xiàn)應(yīng)用重啟時數(shù)據(jù)恢復(fù)。

checkpoint

這里采用的是checkpoint機(jī)制，在重啟或者失敗后重啟可以直接讀取上次沒有完成的任務(wù)，從kafka對應(yīng)offset讀取數(shù)據(jù)。

// 初始化配置文件
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對象，重新統(tǒng)計(jì)計(jì)算pv,uv。

注意：ssc.stop(false,true)表示優(yōu)雅地銷毀StreamingContext對象，不能銷毀SparkContext對象，ssc.stop(true,true)會停掉SparkContext對象，程序就直接停了。

應(yīng)用遷移或者程序升級

在這個過程中，我們把應(yīng)用升級了一下，比如說某個功能寫的不夠完善，或者有邏輯錯誤，這時候都是需要修改代碼，重新打jar包的，這時候如果把程序停了，新的應(yīng)用還是會讀取老的checkpoint，可能會有兩個問題：

1. 執(zhí)行的還是上一次的程序，因?yàn)閏heckpoint里面也有序列化的代碼；
2. 直接執(zhí)行失敗，反序列化失??；

其實(shí)有時候，修改代碼后不用刪除checkpoint也是可以直接生效，經(jīng)過很多測試，我發(fā)現(xiàn)如果對數(shù)據(jù)的過濾操作導(dǎo)致數(shù)據(jù)過濾邏輯改變，還有狀態(tài)操作保存修改，也會導(dǎo)致重啟失敗，只有刪除checkpoint才行，可是實(shí)際中一旦刪除checkpoint，就會導(dǎo)致上一次未完成的任務(wù)和消費(fèi)kafka的offset丟失，直接導(dǎo)致數(shù)據(jù)丟失，這種情況下我一般這么做。

這種情況一般是在另外一個集群，或者把checkpoint目錄修改下，我們是代碼與配置文件分離，所以修改配置文件checkpoint的位置還是很方便的。然后兩個程序一起跑，除了checkpoint目錄不一樣，會重新建，都插入同一個數(shù)據(jù)庫，跑一段時間后，把舊的程序停掉就好。以前看官網(wǎng)這么說，只能記住不能清楚明了，只有自己做時才會想一下辦法去保證數(shù)據(jù)準(zhǔn)確。

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級別的日志會通過郵件發(fā)送到手機(jī)，如果錯誤太多也會被郵件轟炸，需要注意。

val logger = LogManager.getLogger(HelperHandle.getClass.getSimpleName)
  // 郵件level=error日志
  val logger2 = LogManager.getLogger("email")