本文主要是介绍SparkCore（10）：uv/pv实例，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！
1.统计样例
2013-05-19 13:00:00 http://www.taobao.com/17/?tracker_u=1624169&type=1 B58W48U4WKZCJ5D1T3Z9ZY88RU7QA7B1 http://hao.360.cn/ 1.196.34.243 NULL -1
2013-05-19 13:00:00 http://www.taobao.com/item/962967_14?ref=1_1_52_search.ctg_1 T82C9WBFB1N8EW14YF2E2GY8AC9K5M5P http://www.yihaodian.com/ctg/s2/c24566-%E5%B1%B1%E6%A5%82%E5%88%B6%E5%93%81?ref=pms_15_78_258 222.78.246.228 134939954 156
2013-05-19 13:00:00 http://www.taobao.com/1/?tracker_u=1013304189&uid=2687512&type=3 W17C89RU8DZ6NMN7JD2ZCBDMX1CQVZ1W http://www.yihaodian.com/1/?tracker_u=1013304189&uid=2687512&type=3 118.205.0.18 NULL -202.代码
2.1 SparkUtil
package SparkUtilimport org.apache.spark.{SparkConf, SparkContext}/*** Created by ibf on 2018/7/18.*/
object SparkUtil {def createSparkContext(isLocal:Boolean,appName:String): SparkContext ={if(isLocal) {val conf = new SparkConf().setAppName(appName).setMaster("local[2]")val sc = SparkContext.getOrCreate(conf)val ssc=SparkContext.getOrCreate(conf)sc}else{val conf = new SparkConf().setAppName(appName)val sc = SparkContext.getOrCreate(conf)sc}}}
2.2 SparkPVAndUV
package _0722rddimport SparkUtil.SparkUtil
import org.apache.spark.rdd.RDD/*** */
object SparkPVAndUV {def main(args: Array[String]) {val sc = SparkUtil.createSparkContext(true,"SparkPVAndUV")
// val path = "hdfs://192.168.244.101:8020/page_views.data"val path = "hdfs://192.168.31.3:8020/page_views.data"val originalRdd: RDD[String] = sc.textFile(path)//因为缓存不是立即操作的api，只有当调用了这块缓存的数据才会cacheoriginalRdd.cache()//originalRdd.count()//某些固定的值，应该要写在配置中，然后通过读取配置来获取val arrLen = 7val timeLen = 16//处理过后的rddval mappedRdd: RDD[(String, String, String)] = originalRdd.map(_.split("\t")).filter(arr =>{arr.length == arrLen && arr(0).trim.length > timeLen && arr(1).length > 0}).map(arr =>{//每分钟的pvval date = arr(0).substring(0,16)val url = arr(1).trimval guid = arr(2).trim(date,url,guid)})mappedRdd.cache()mappedRdd.count()//一、计算PV/*** 其实计算pv只要维度（date）和url*///XXXByKey的操作是针对于PairRdd（二元组rdd）才能实现的，val resultRdd = mappedRdd.map(t => (t._1,t._2)).groupByKey().map {//date就是日期，itr是迭代器，里面把相同日期的value全部放到一起case (date, itr) => {(date, itr.size)}}//resultRdd结果：Array[(String, Int)] = Array((2013-05-19 13:35,3504))// resultRdd.foreach(println)//思考：groupByKey这样的API，有没有什么其他API可以实现这个功能，他们之间的性能比较//这段代码有哪些地方是可以优化的/*** 优化groupByKey： grouByKey 这个api性能不是特别好* 会把相同key的所有数据全部放到同一个迭代器中，数据倾斜* API可以替换，* 是否可以不保留url的值，直接写1，然后用于后面的count*///def reduceByKey(func: (V, V) => V, numPartitions: Int)/*** 这里有一个numPartitions可以指定，分区数量* executor 5 个core 就可以并行计算5个分区的数据* 当数量大的时候，甚至出现数据倾斜的时候，可以通过增加分区数量来缓解每个task的计算压力*/val pvRdd = mappedRdd.map(t => (t._1,1)) //mappedRdd.map(t => (t._1,1))为Array[(String, Int)] = Array((2013-05-19 13:00,1), (2013-05-19 13:00,1)).reduceByKey(_ + _,5)pvRdd.foreach(println) //结果是Array((2013-05-19 13:07,3486), (2013-05-19 13:16,3395))Thread.sleep(100000l)originalRdd.unpersist()mappedRdd.unpersist()//=================================================================================================//二、计算uv/*** uv应该如何计算？count distinct groupby(XXX,xxx)* select count(distinct XXX) as uv from XX group by XXX* select count(1) from (select XXX from group by xxx ) tb* 在什么场景下应该用哪一种呢* key较为分散的情况下使用groupByKey， key较为集中的情况下使用reduceByKey*//**方法一**//* val uvRdd = mappedRdd.filter(t => t._3.nonEmpty).map(t => {//把什么作为key然后进行聚合，每分钟的uv(t._1,t._3)}).groupByKey().map({case (date,itr) =>{(date,itr.toSet.size)}})*//**方法二：是否可以使用reduceByKey来做去重呢？* 我只想知道在同一个时间段内出现了多少key，key出现的次数，并不不关注* spark rdd的api的时候，要关注，你的key是什么？*///(date,url,uid)val uvRdd = mappedRdd.filter(t => t._3.nonEmpty)//((date,uid),1) 下面这个是去重操作.map(t => ((t._1,t._3),1)).reduceByKey({case (a,b) => a})//进行第二次聚合 ((13:01,uid1),1),((13:01,uid2),1),((13:01,uid3),1)//想要得到(13:01,3).map({case ((date,uid),int) =>{(date,1)}}).reduceByKey(_ + _)/*** 方法三：spark常用去重API*//* val uvRdd = mappedRdd.filter(t => t._3.nonEmpty).map(t => (t._1,t._3)).distinct(10).map(t =>(t._1,1)).reduceByKey(_ + _)*///uvRdd.foreach(println)//使用外联，计算出值的就保留值，没计算出来的就给定默认值-1/*** select date,* (case when pv is not null* then pv* else* -1) as pv,* (case when uv is not null* then uv* else* -1) as uv from (select date,pv from A full join B on A.date = B.date) tb*/
// val resultRdd: RDD[(String, Int, Int)] = pvRdd.fullOuterJoin(uvRdd)
// .map({
// case(date,(optpv,optuv)) =>{
// (date,optpv.getOrElse(-1),optuv.getOrElse(-1))
// }
// }).coalesce(1)
// resultRdd.foreach(println)//==================================================================================
//================输出==============================================================
// resultRdd.foreach(println)
// resultRdd.saveAsTextFile(s"hdfs://192.168.244.101:8020/" +
// s"spark/sparkPVUV_${System.currentTimeMillis()}")// Thread.sleep(100000l)// originalRdd.unpersist()
// mappedRdd.unpersist()Thread.sleep(100000000l)}
}
这篇关于SparkCore（10）：uv/pv实例的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！
