最近有个需求，实时统计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一样，并且性能更好。当然了实际中数据过来需要经过清洗，过滤，才能使用。
定义一个状态函数
//实时流量状态更新函数valmapFunction=(datehour:String,pv:Option[Long],state:State[Long])=>{valaccuSum=pv.getOrElse(0L)+state.getOption().getOrElse(0L)valoutput=(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连接valjedis=getJediseachPartition.foreach(x=>{//省略若干…jedis.sadd(key,x._2)//设置存储每天的数据的set过期时间，防止超过redis容量，这样每天的set集合，定期会被自动删除jedis.expire(key,ConfigFactory.rediskeyexists)})//关闭连接closeJedis(jedis)})}) 3)结果保存到数据库
结果保存到mysql，数据库，10秒刷新一次数据库，前端展示刷新一次，就会重新查询一次数据库，做到实时统计展示pv,uv的目的。
/***插入数据*@paramdata(addTab(datehour)+helperversion)*@paramtbName*@paramcolNames*/definsertHelper(data:DStream[(String,Long)],tbName:String,colNames:String*):Unit={data.foreachRDD(rdd=>{valtmp_rdd=rdd.map(x=>x._1.substring(11,13).toInt)if(!rdd.isEmpty()){valhour_now=tmp_rdd.max()//获取当前结果中最大的时间，在数据恢复中可以起作用rdd.foreachPartition(eachPartition=>{try{valjedis=getJedisvalconn=MysqlPoolUtil.getConnection()conn.setAutoCommit(false)valstmt=conn.createStatement()eachPartition.foreach(x=>{//valsql=….//省略若干stmt.addBatch(sql)})closeJedis(jedis)stmt.executeBatch()//批量执行sql语句conn.commit()conn.close()}catch{casee:Exception=>{logger.error(e)logger2.error(HelperHandle.getClass.getSimpleName+e)}}})}})}//计算当前时间距离次日零点的时长（毫秒）defresetTime={valnow=newDate()valtodayEnd=Calendar.getInstancetodayEnd.set(Calendar.HOUR_OF_DAY,23)//Calendar.HOUR12小时制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()valconf=newSparkConf().setAppName(ConfigFactory.sparkstreamname)conf.set(\”spark.streaming.stopGracefullyOnShutdown\”,\”true\”)conf.set(\”spark.streaming.kafka.maxRatePerPartition\”,consumeRate)conf.set(\”spark.default.parallelism\”,\”24\”)valsc=newSparkContext(conf)while(true){valssc=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语义。
varmessages:InputDStream[ConsumerRecord[String,String]]=nullif(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**@paramtbName*@return*/defgetLastOffsets(tbName:String):mutable.HashMap[TopicPartition,Long]={valsql=s\”selectoffsetfrom${tbName}whereid=(selectmax(id)from${tbName})\”valconn=MysqlPool.getConnection(config)valpsts=conn.prepareStatement(sql)valres=psts.executeQuery()vartpMap:mutable.HashMap[TopicPartition,Long]=mutable.HashMap[TopicPartition,Long]()while(res.next()){valo=res.getString(1)valjSONArray=JSON.parseArray(o)jSONArray.toArray().foreach(offset=>{valjson=JSON.parseObject(offset.toString)valtopicAndPartition=newTopicPartition(json.getString(\”topic\”),json.getInteger(\”partition\”))tpMap.put(topicAndPartition,json.getLong(\”untilOffset\”))})}MysqlPool.closeCon(res,psts,conn)tpMap} 6)日志
日志用的log4j2，本地保存一份，ERROR级别的日志会通过邮件发送到手机，如果错误太多也会被邮件轰炸，需要注意。
vallogger=LogManager.getLogger(HelperHandle.getClass.getSimpleName)//邮件level=error日志vallogger2=LogManager.getLogger(\”email\”)
