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

最近有個需求,實時統計pv,uv,結果按照date,hour,pv,uv來展示,按天統計,第二天重新統計,當然了實際還需要按照類型字段分類統計pv,uv,比如按照date,hour,pv,uv,type來展示。這里介紹最基本的pv,uv的展示。

1、項目流程


日志數據從flume采集過來,落到hdfs供其它離線業務使用,也會sinkkafka,sparkStreamingkafka拉數據過來,計算pv,uv,uv是用的redisset集合去重,最后把結果寫入mysql數據庫,供前端展示使用。

2、具體過程

1)pv的計算

拉取數據有兩種方式,基于receiveddirect方式,這里用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
  }
Java
 計算pv
 val stateSpec = StateSpec.function(mapFunction)
 val helper_count_all = helper_data.map(x => (x._1,1L)).mapWithState(stateSpec).stateSnapshots().repartition(2)
Java

這樣就很容易的把pv計算出來了。

2)uv的計算

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

helper_data_dis.foreachRDD(rdd => {
  rdd.foreachPartition(eachPartition => {
    var jedis: Jedis = null
    try {
      jedis = getJedis
      eachPartition.foreach(x => {
        val arr = x._2.split("\t")
        val date: String = arr(0).split(":")(0)

        // helper 統計
        val key0 = "helper_" + date
        jedis.sadd(key0, x._1)
        jedis.expire(key0, ConfigFactory.rediskeyexists)
        // helperversion 統計
        val key = date + "_" + arr(1)
        jedis.sadd(key, x._1)
        jedis.expire(key, ConfigFactory.rediskeyexists)
      })
    } catch {
      case e: Exception => {
        logger.error(e)
        logger2.error(HelperHandle.getClass.getSimpleName + e)
      }
    } finally {
      if (jedis != null) {
        closeJedis(jedis)
      }
    }
  })
})

// 獲取jedis連接
def getJedis: Jedis = {
  val jedis = RedisPoolUtil.getPool.getResource
  jedis
}

// 釋放jedis連接
def closeJedis(jedis: Jedis): Unit = {
  RedisPoolUtil.getPool.returnResource(jedis)
}
Java

redis連接池代碼RedisPoolUtil.scala

package com.js.ipflow.utils

import com.js.ipflow.start.ConfigFactory
import org.apache.commons.pool2.impl.GenericObjectPoolConfig
import redis.clients.jedis.JedisPool

/**
  * redis 連接池工具類
  * @author keguang
  */

object RedisPoolUtil extends Serializable{
  @transient private var pool: JedisPool = null

  /**
    * 讀取jedis配置信息, 出發jedis初始化
    */
  def initJedis: Unit ={
    ConfigFactory.initConfig()
    val maxTotal = 50
    val maxIdle = 30
    val minIdle = 10
    val redisHost = ConfigFactory.redishost
    val redisPort = ConfigFactory.redisport
    val redisTimeout = ConfigFactory.redistimeout
    val redisPassword = ConfigFactory.redispassword
    makePool(redisHost, redisPort, redisTimeout, redisPassword, maxTotal, maxIdle, minIdle)
  }

  def makePool(redisHost: String, redisPort: Int, redisTimeout: Int,redisPassword:String, maxTotal: Int, maxIdle: Int, minIdle: Int): Unit = {
   init(redisHost, redisPort, redisTimeout, redisPassword, maxTotal, maxIdle, minIdle, true, false, 10000)
  }

  /**
    * 初始化jedis連接池
    * @param redisHost host
    * @param redisPort 端口
    * @param redisTimeout 連接redis超時時間
    * @param redisPassword redis密碼
    * @param maxTotal 總的連接數
    * @param maxIdle 最大空閑連接數
    * @param minIdle 最小空閑連接數
    * @param testOnBorrow
    * @param testOnReturn
    * @param maxWaitMillis
    */
  def init(redisHost: String, redisPort: Int, redisTimeout: Int,redisPassword:String, maxTotal: Int, maxIdle: Int, minIdle: Int, testOnBorrow: Boolean, testOnReturn: Boolean, maxWaitMillis: Long): Unit = {
    if (pool == null) {
      val poolConfig = new GenericObjectPoolConfig()
      poolConfig.setMaxTotal(maxTotal)
      poolConfig.setMaxIdle(maxIdle)
      poolConfig.setMinIdle(minIdle)
      poolConfig.setTestOnBorrow(testOnBorrow)
      poolConfig.setTestOnReturn(testOnReturn)
      poolConfig.setMaxWaitMillis(maxWaitMillis)
      pool = new JedisPool(poolConfig, redisHost, redisPort, redisTimeout,redisPassword)

      val hook = new Thread {
        override def run = pool.destroy()
      }
      sys.addShutdownHook(hook.run)
    }
  }

  def getPool: JedisPool = {
    if(pool == null){
      initJedis
    }
    pool
  }

}
Java

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 => {
        var jedis: Jedis = null
        var conn: Connection = null
        try {
          jedis = getJedis
          conn = MysqlPoolUtil.getConnection()
          conn.setAutoCommit(false)
          val stmt = conn.createStatement()
          eachPartition.foreach(x => {
            if (colNames.length == 7) {
              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},${colName6}) values(${colValue0},${colValue1},${colValue2},${colValue3},${colValue4},${colValue5},${colValue6}) on duplicate key update ${colName2} =  ${colValue2},${colName3} = ${colValue3}"
                logger.warn(sql)
                stmt.addBatch(sql)
              } /* else {
              sql = s"insert into ${tbName}(${colName0},${colName1},${colName2},${colName4},${colName5},${colName6}) values(${colValue0},${colValue1},${colValue2},${colValue4},${colValue5},${colValue6}) on duplicate key update ${colName2} =  ${colValue2}"
            }*/
            } else if (colNames.length == 5) {
              val date_hour = x._1.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) // helper_count_all
              val colName3 = colNames(3) // helper_count
              val colName4 = colNames(4) // dh

              val colValue0 = addYin(date)
              val colValue1 = hour
              val colValue2 = x._2.toInt
              val colValue3 = jedis.scard("helper_" + date) // // helper_2018-07-08
              val colValue4 = 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}) values(${colValue0},${colValue1},${colValue2},${colValue3},${colValue4}) on duplicate key update ${colName2} =  ${colValue2},${colName3} = ${colValue3}"
                logger.warn(sql)
                stmt.addBatch(sql)
              }
            }
          })
          stmt.executeBatch() // 批量執行sql語句
          conn.commit()
        } catch {
          case e: Exception => {
            logger.error(e)
            logger2.error(HelperHandle.getClass.getSimpleName + e)
          }
        } finally {
          if (jedis != null) {
            closeJedis(jedis)
          }

          if(conn != null){
            conn.close()
          }
        }
      })
    }
  })
}

// 計算當前時間距離次日零點的時長(毫秒)
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
 }
Java

msql 連接池代碼MysqlPoolUtil.scala

package com.js.ipflow.utils

import java.sql.{Connection, PreparedStatement, ResultSet}

import com.js.ipflow.start.ConfigFactory
import org.apache.commons.dbcp.BasicDataSource
import org.apache.logging.log4j.LogManager

/**
  *jdbc mysql 連接池工具類
  * @author keguang
  */
object MysqlPoolUtil {

  val logger = LogManager.getLogger(MysqlPoolUtil.getClass.getSimpleName)

  private var bs:BasicDataSource = null

  /**
    * 創建數據源
    * @return
    */
  def getDataSource():BasicDataSource={
    if(bs==null){
      ConfigFactory.initConfig()
      bs = new BasicDataSource()
      bs.setDriverClassName("com.mysql.jdbc.Driver")
      bs.setUrl(ConfigFactory.mysqlurl)
      bs.setUsername(ConfigFactory.mysqlusername)
      bs.setPassword(ConfigFactory.mysqlpassword)
      bs.setMaxActive(50)           // 設置最大并發數
      bs.setInitialSize(20)          // 數據庫初始化時,創建的連接個數
      bs.setMinIdle(20)              // 在不新建連接的條件下,池中保持空閑的最少連接數。
      bs.setMaxIdle(20)             // 池里不會被釋放的最多空閑連接數量。設置為0時表示無限制。
      bs.setMaxWait(5000)             // 在拋出異常之前,池等待連接被回收的最長時間(當沒有可用連接時)。設置為-1表示無限等待。
      bs.setMinEvictableIdleTimeMillis(10*1000)     // 空閑連接5秒中后釋放
      bs.setTimeBetweenEvictionRunsMillis(1*60*1000)      //1分鐘檢測一次是否有死掉的線程
      bs.setTestOnBorrow(true)
    }
    bs
  }

  /**
    * 釋放數據源
    */
  def shutDownDataSource(){
    if(bs!=null){
      bs.close()
    }
  }

  /**
    * 獲取數據庫連接
    * @return
    */
  def getConnection():Connection={
    var con:Connection = null
    try {
      if(bs!=null){
        con = bs.getConnection()
      }else{
        con = getDataSource().getConnection()
      }
    } catch{
      case e:Exception => logger.error(e)
    }
    con
  }

  /**
    * 關閉連接
    */
  def closeCon(rs:ResultSet ,ps:PreparedStatement,con:Connection){
    if(rs!=null){
      try {
        rs.close()
      } catch{
        case e:Exception => println(e.getMessage)
      }
    }
    if(ps!=null){
      try {
        ps.close()
      } catch{
        case e:Exception => println(e.getMessage)
      }
    }
    if(con!=null){
      try {
        con.close()
      } catch{
        case e:Exception => println(e.getMessage)
      }
    }
  }
}
Java

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)
}
Java






checkpoint是每天一個目錄,在第二天凌晨定時銷毀StreamingContext對象,重新統計計算pv,uv。

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

應用遷移或者程序升級

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

  1. 執行的還是上一次的程序,因為checkpoint里面也有序列化的代碼;
  2. 直接執行失敗,反序列化失??;

其實有時候,修改代碼后不用刪除checkpoint也是可以直接生效,經過很多測試,我發現如果對數據的過濾操作導致數據過濾邏輯改變,還有狀態操作保存修改,也會導致重啟失敗,只有刪除checkpoint才行,可是實際中一旦刪除checkpoint,就會導致上一次未完成的任務和消費kafkaoffset丟失,直接導致數據丟失,這種情況下我一般這么做。

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

5)日志

日志用的log4j2,本地保存一份,ERROR級別的日志會通過郵件發送到郵箱。

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

3、主要代碼

需要的maven依賴:

        
            org.apache.spark
            spark-core_2.11
            ${spark.version}
            provided
        
        
            org.apache.spark
            spark-streaming_2.11
            ${spark.version}
            provided
        
        
            mysql
            mysql-connector-java
            5.1.40
        
        
            commons-dbcp
            commons-dbcp
            1.4
            provided
        
XML

讀取配置文件代碼ConfigFactory .java

package com.js.ipflow.start;

import com.google.common.io.Resources;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.dom4j.Document;
import org.dom4j.DocumentException;
import org.dom4j.Element;
import org.dom4j.io.SAXReader;

import java.io.File;

public class ConfigFactory {
    private final static Logger log = LogManager.getLogger("email");

    public static String kafkaipport;
    public static String kafkazookeeper;
    public static String kafkatopic;
    public static String kafkagroupid;
    public static String mysqlurl;
    public static String mysqlusername;
    public static String mysqlpassword;
    public static String redishost;
    public static int redisport;
    public static String redispassword;
    public static int redistimeout;
    public static int rediskeyexists;
    public static String sparkstreamname;
    public static int sparkstreamseconds;
    public static String sparkstreammaster = "spark://qcloud-spark01:7077"; // 僅供本地測試使用
    public static String localpath;
    public static String checkpointdir;
    // public static String gracestopfile; // 優雅得kill掉程序
    public static String keydeserilizer;
    public static String valuedeserilizer;

    /**
     * 初始化所有的通用信息
     */
    public static void initConfig(){readCommons();}

    /**
     * 讀取commons.xml文件
     */
    private static void readCommons(){
        SAXReader reader = new SAXReader(); // 構建xml解析器
        Document document = null;
        try{
            document = reader.read(Resources.getResource("commons.xml"));
        }catch (DocumentException e){
            log.error("ConfigFactory.readCommons",e);
        }

        if(document != null){
            Element root = document.getRootElement();

            Element kafkaElement = root.element("kafka");
            kafkaipport = kafkaElement.element("ipport").getText();
            kafkazookeeper = kafkaElement.element("zookeeper").getText();
            kafkatopic = kafkaElement.element("topic").getText();
            kafkagroupid = kafkaElement.element("groupid").getText();
            keydeserilizer=kafkaElement.element("keySer").getText();
            valuedeserilizer=kafkaElement.element("valSer").getText();

            Element mysqlElement = root.element("mysql");
            mysqlurl = mysqlElement.element("url").getText();
            mysqlusername = mysqlElement.element("username").getText();
            mysqlpassword = mysqlElement.element("password").getText();

            Element redisElement = root.element("redis");
            redishost = redisElement.element("host").getText();
            redisport = Integer.valueOf(redisElement.element("port").getText());
            redispassword = redisElement.element("password").getText();
            redistimeout = Integer.valueOf(redisElement.element("timeout").getText());
            rediskeyexists = Integer.valueOf(redisElement.element("keyexists").getText());

            Element sparkElement = root.element("spark");
            // sparkstreammaster = sparkElement.element("streammaster").getText();
            sparkstreamname = sparkElement.element("streamname").getText();
            sparkstreamseconds = Integer.valueOf(sparkElement.element("seconds").getText());

            Element pathElement = root.element("path");
            localpath = pathElement.element("localpath").getText();
            checkpointdir = pathElement.element("checkpointdir").getText();
            // gracestopfile = pathElement.element("gracestopfile").getText();

        }else {
            log.warn("commons.xml配置文件讀取錯誤...");
        }
    }
}
Java

主要業務代碼,如下:

package com.js.ipflow.flash.helper

import java.sql.Connection
import java.util.{Calendar, Date}

import com.alibaba.fastjson.JSON
import com.js.ipflow.start.ConfigFactory
import com.js.ipflow.utils.{DateUtil, MysqlPoolUtil, RedisPoolUtil}
import kafka.serializer.StringDecoder
import org.apache.logging.log4j.LogManager
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, State, StateSpec, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}
import redis.clients.jedis.Jedis

object HelperHandle {

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

  def main(args: Array[String]): Unit = {
    helperHandle(args(0))
  }

  def helperHandle(consumeRate: String): Unit = {

    // 初始化配置文件
    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", "30")
    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)
    }

    def getStreamingContext(): StreamingContext = {
      val stateSpec = StateSpec.function(mapFunction)
      val ssc = new StreamingContext(sc, Seconds(ConfigFactory.sparkstreamseconds))
      ssc.checkpoint(ConfigFactory.checkpointdir + DateUtil.getDay(0))
      val zkQuorm = ConfigFactory.kafkazookeeper
      val topics = ConfigFactory.kafkatopic
      val topicSet = Set(topics)
      val kafkaParams = Map[String, String](
        "metadata.broker.list" -> (ConfigFactory.kafkaipport)
        , "group.id" -> (ConfigFactory.kafkagroupid)
        , "auto.offset.reset" -> kafka.api.OffsetRequest.LargestTimeString
      )

      val rmessage = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
        ssc, kafkaParams, topicSet
      )

      // helper數據 (dateHour,guid,helperversion)
      val helper_data = FilterHelper.getHelperData(rmessage.map(x => {
        val message = JSON.parseObject(x._2).getString("message")
        JSON.parseObject(message)
      })).repartition(60).cache()

      // (guid, datehour + helperversion)
      val helper_data_dis = helper_data.map(x => (x._2, addTab(x._1) + x._3)).reduceByKey((x, y) => y)

      // pv,uv
      val helper_count = helper_data.map(x => (x._1, 1L)).mapWithState(stateSpec).stateSnapshots().repartition(2)

      // helperversion
      val helper_helperversion_count = helper_data.map(x => (addTab(x._1) + x._3, 1L)).mapWithState(stateSpec).stateSnapshots().repartition(2)
      helper_data_dis.foreachRDD(rdd => {
        rdd.foreachPartition(eachPartition => {
          var jedis: Jedis = null
          try {
            jedis = getJedis
            eachPartition.foreach(x => {
              val arr = x._2.split("\t")
              val date: String = arr(0).split(":")(0)

              // helper 統計
              val key0 = "helper_" + date
              jedis.sadd(key0, x._1)
              jedis.expire(key0, ConfigFactory.rediskeyexists)
              // helperversion 統計
              val key = date + "_" + arr(1)
              jedis.sadd(key, x._1)
              jedis.expire(key, ConfigFactory.rediskeyexists)
            })
          } catch {
            case e: Exception => {
              logger.error(e)
              logger2.error(HelperHandle.getClass.getSimpleName + e)
            }
          } finally {
            if (jedis != null) {
              closeJedis(jedis)
            }
          }
        })
      })
      insertHelper(helper_helperversion_count, "statistic_realtime_flash_helper", "date", "hour", "count_all", "count", "helperversion", "datehour", "dh")
      insertHelper(helper_count, "statistic_realtime_helper_count", "date", "hour", "helper_count_all", "helper_count", "dh")

      ssc
    }
  }

  ///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
  // 計算當前時間距離次日零點的時長(毫秒)
  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
  }

  /**
    * 插入數據
    *
    * @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 => {
          var jedis: Jedis = null
          var conn: Connection = null
          try {
            jedis = getJedis
            conn = MysqlPoolUtil.getConnection()
            conn.setAutoCommit(false)
            val stmt = conn.createStatement()
            eachPartition.foreach(x => {
              if (colNames.length == 7) {
                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},${colName6}) values(${colValue0},${colValue1},${colValue2},${colValue3},${colValue4},${colValue5},${colValue6}) on duplicate key update ${colName2} =  ${colValue2},${colName3} = ${colValue3}"
                  logger.warn(sql)
                  stmt.addBatch(sql)
                } /* else {
                sql = s"insert into ${tbName}(${colName0},${colName1},${colName2},${colName4},${colName5},${colName6}) values(${colValue0},${colValue1},${colValue2},${colValue4},${colValue5},${colValue6}) on duplicate key update ${colName2} =  ${colValue2}"
              }*/
              } else if (colNames.length == 5) {
                val date_hour = x._1.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) // helper_count_all
                val colName3 = colNames(3) // helper_count
                val colName4 = colNames(4) // dh

                val colValue0 = addYin(date)
                val colValue1 = hour
                val colValue2 = x._2.toInt
                val colValue3 = jedis.scard("helper_" + date) // // helper_2018-07-08
                val colValue4 = 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}) values(${colValue0},${colValue1},${colValue2},${colValue3},${colValue4}) on duplicate key update ${colName2} =  ${colValue2},${colName3} = ${colValue3}"
                  logger.warn(sql)
                  stmt.addBatch(sql)
                }
              }
            })
            stmt.executeBatch() // 批量執行sql語句
            conn.commit()
          } catch {
            case e: Exception => {
              logger.error(e)
              logger2.error(HelperHandle.getClass.getSimpleName + e)
            }
          } finally {
            if (jedis != null) {
              closeJedis(jedis)
            }

            if(conn != null){
              conn.close()
            }
          }
        })
      }
    })
  }

  def addYin(str: String): String = {
    "'" + str + "'"
  }

  // 字符串添加tab格式化方法
  def addTab(str: String): String = {
    str + "\t";
  }

  // 實時流量狀態更新函數
  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
  }

  // 獲取jedis連接
  def getJedis: Jedis = {
    val jedis = RedisPoolUtil.getPool.getResource
    jedis
  }

  // 釋放jedis連接
  def closeJedis(jedis: Jedis): Unit = {
    RedisPoolUtil.getPool.returnResource(jedis)
  }

}








作者:柯廣的網絡日志

微信公眾號:Java大數據與數據倉庫