實時統計每天pv,uv的sparkStreaming結合redis結果存入mysql供前端展示
- 2019 年 10 月 3 日
- 筆記
最近有個需求，實時統計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
1、項目流程
日誌數據從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 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，可能會有兩個問題：
- 執行的還是上一次的程式，因為checkpoint裡面也有序列化的程式碼；
- 直接執行失敗，反序列化失敗；
其實有時候，修改程式碼後不用刪除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") 分享一個大神的人工智慧教程。零基礎！通俗易懂！風趣幽默！還帶黃段子！希望你也加入到人工智慧的隊伍中來！
微信公眾號
我的微信公眾號，專註於大數據分析與挖掘，感興趣可以關注，看一看，瞧一瞧！
