實時統計天天pv,uv的sparkStreaming結合redis結果存入mysql供前端展現

最近有個需求，實時統計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，能夠參見這篇博客:https://blog.csdn.net/petermsh/article/details/78652246

一、項目流程

日誌數據從flume採集過來，落到hdfs供其它離線業務使用，也會sink到kafka，sparkStreaming從kafka拉數據過來，計算pv,uv，uv是用的redis的set集合去重，最後把結果寫入mysql數據庫，供前端展現使用。

二、具體過程

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
 val stateSpec = StateSpec.function(mapFunction)
 val helper_count_all = helper_data.map(x => (x._1,1L)).mapWithState(stateSpec).stateSnapshots().repartition(2)

這樣就很容易的把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 => {
            val date:String = x._1.split(":")(0)
            val key = date
            // 將date做爲key，guid(x._2)加入set集合
            jedis.sadd(key,x._2)
            // 設置存儲天天的數據的set過時時間，防止超過redis容量，這樣天天的set集合，按期會被自動刪除
            jedis.expire(key,ConfigFactory.rediskeyexists)
          })
          // 關閉鏈接
          closeJedis(jedis)
        })
      })

3）結果保存到數據庫

結果保存到mysql，數據庫，20秒刷新一次數據庫，前端展現刷新一次，就會從新查詢一次數據庫，作到實時統計展現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 datehour = x._1.split("\t")(0)
              val helperversion = x._1.split("\t")(1)
              val date_hour = datehour.split(":")
              val date = date_hour(0)
              val hour = date_hour(1).toInt

              val colName0 = colNames(0) // date
              val colName1 = colNames(1) // hour
              val colName2 = colNames(2) // count_all
              val colName3 = colNames(3) // count
              val colName4 = colNames(4) // helperversion
              val colName5 = colNames(5) // datehour
              val colName6 = colNames(6) // dh

              val colValue0 = addYin(date)
              val colValue1 = hour
              val colValue2 = x._2.toInt
              val colValue3 = jedis.scard(date + "_" + helperversion) // // 2018-07-08_10.0.1.22
              val colValue4 = addYin(helperversion)
              var colValue5 = if (hour < 10) "'" + date + " 0" + hour + ":00 " + helperversion + "'" else "'" + date + " " + hour + ":00 " + helperversion + "'"
              val colValue6 = if(hour < 10) "'" + date + " 0" + hour + ":00'" else "'" + date + " " + hour + ":00'"

              var sql = ""
              if (hour == hour_now) { // uv只對如今更新
                sql = s"insert into ${tbName}(${colName0},${colName1},${colName2},${colName3},${colName4},${colName5}) values(${colValue0},${colValue1},${colValue2},${colValue3},${colValue4},${colValue5}) on duplicate key update ${colName2} =  ${colValue2},${colName3} = ${colValue3}"
              } else {
                sql = s"insert into ${tbName}(${colName0},${colName1},${colName2},${colName4},${colName5}) values(${colValue0},${colValue1},${colValue2},${colValue4},${colValue5}) on duplicate key update ${colName2} =  ${colValue2}"
              }
              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）日誌

日誌用的log4j2，本地保存一份，ERROR級別的日誌會經過郵件發送到手機。

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

---

分享一個大神的人工智能教程。零基礎！通俗易懂！風趣幽默！還帶黃段子！但願你也加入到人工智能的隊伍中來！

點擊瀏覽教程

微信公衆號

個人微信公衆號，專一於大數據分析與挖掘，感興趣能夠關注，看一看，瞧一瞧！