关注公众号：分享电脑学习
回复"百度云盘" 可以免费获取所有学习文档的代码（不定期更新）
云盘目录说明：
tools目录是安装包
res 目录是每一个课件对应的代码和资源等
doc 目录是一些第三方的文档工具
承接上一篇文档《Spark应用的结构》
参数说明：
创建一个Maven项目
Pom文件引入jar，配置信息已经完成，大家可以直接去网盘下载，版本号根据自己的安装情况调整
部分示例
创建一个scala文件，编写代码
1. 构建SparkContext上下文对象
valconf =newSparkConf()
valsc =newSparkContext(conf)
此时可以运行这两行代码
会出现错误
需要添加一行代码
setMaster("local")// 指定应用在哪儿执行，可以是local、或者stadnalone、yarn、mesos集群
再运行一次
又报一次错，这个是要求指定应用的名字
添加代码
.setAppName("pvtest")//指定应用的名字
再运行就可以了（启动hadoop）
2. 基于sc构建RDD
端口为core-site.xml中配置的
将文件上传到指定目录
valpath="hdfs://ip:8020/data/page_views.data"//HDFS的schema 给定数据的路径
val rdd: RDD[String] = sc.textFile(path)
println("总共有"+ rdd.count()+"条数据")
运行可以看到显示
如果不想写schema的话，需要将hadoop的两个文件拷贝到项目的src/main/resources中
两个文件是hadoop的：hdfs-site.xml和core-site.xml
编写代码
valpath = /data/page_views.data
运行也可以查看到结果
3. 业务实现
思路：
（1）分析可知道：数据分为7个字段，业务需要三个字段（时间，URL，guid），计算某一个时间的PV的值
（2）数据进行过滤清洗，获取两个字段（时间、url）
（3）url非空，时间非空，时间字符串的长度必须大于10
（4）sql:selectdate,count(url)frompage_viewgroupbydate;
（5）sql:selectdate,count(1)frompage_viewgroupbydate;
（6）分别用reduceByKey和groupByKey进行数据处理
我们一步步来
先分割数据
val rdd1 = rdd.map(line=>line.split("\t"))
数据进行过滤清洗，获取两个字段（时间、url）
url非空，时间非空，时间字符串的长度必须大于10
.filter(arr=>{
//保留正常数据
arr.length >2 && arr(1).trim.nonEmpty && arr(0).trim.length > 10
})
截取数据
.map(arr=>{
val date = arr(0).trim.substring(0,10)
val url = arr(1).trim
(date,1)// (date,url)
})
基于reduceByKey做统计pv
val pvRdd = rdd1.reduceByKey(_+_)
println("pv------------------"+ pvRdd.collect().mkString(";"))
数据表示2013年5月19日一共有100000条访问数据
也可以基于groupByKey实现pv统计（这个可以试一下，如果不行就使用reduceByKey）
groupByKey相当于把相同的key的value放到迭代器里面，也就是这些value都放到内存里面，如果value值数据量撑爆内存，就会OOM异常
val pvRdd = rdd1.groupByKey()
.map(t=>{
val date = t._1
val pv = t._2.size
(date,pv)
})
println("pv------------------"+ pvRdd.collect().mkString(";"))
与上面值相同
网友评论