SparkStreaming項目實戰，實時計算Pv和Uv
本文轉載自微信公眾號「Java大數據與數據倉庫」，作者柯少爺。轉載本文請聯系Java大數據與數據倉庫公眾號。
最近有個需求，實時統計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計算出來了。
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，可能會有兩個問題：
執行的還是上一次的程序，因為checkpoint里面也有序列化的代碼;
直接執行失敗，反序列化失敗;
其實有時候，修改代碼后不用刪除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")
