文章目录

• 创建一个Maven项目，添加依赖
• 词频统计案例开发及上传jar包到服务器并准备测试数据
• jar包提交Spark应用程序运行
• 词频统计案例迭代之输出结果到HDFS
• 词频统计案例迭代之处理多个输入文件
• 词频统计案例迭代之输入文件规则匹配
• 带排序的词频统计案例开发及运行过程深度剖析
• 求用户访问量的TopN的Hive实现以及Spark Core实现过程分析
• 求平均年龄的Spark Core实现
• 求男女人数以及最高和最低身高

spark-shell适合做测试
如果是开发的话是用IDEA+Maven+Scala来开发Spark应用程序
创建一个Maven项目，添加依赖 前面创建的过程省略。
创建了一个Scala的Maven项目后，需要添加依赖什么的。
我这边pom.xml文件添加代码如下：

2.11.8 2.6.0-cdh5.7.0 2.4.0 org.apache.spark spark-core_2.11 ${spark.version} org.scala-lang scala-library ${scala.version} org.apache.hadoop hadoop-client ${hadoop.version}

添加之后，点击Reimport，重新导入一下

文章图片

但是导入之后，Hadoop的有些没有导进来，红色波浪线，报错了：

文章图片

需要加一个仓库repository：
http://repository.cloudera.com/artifactory/cloudera-repos/

文章图片

继续点击到如下目录：
http://repository.cloudera.com/artifactory/cloudera-repos/org/apache/hadoop/hadoop-client/2.6.0-cdh5.7.0/
这个就是上面我们添加的2.6.0-cdh5.7.0/
如果写的是2.6.0-cdh5.7.0，就要加这个仓库，如果写的是2.6.0，不说cdh的就不用加这个仓库。

文章图片

然后再pom.xml里添加一下

cloudera cloudera http://repository.cloudera.com/artifactory/cloudera-repos/

再重新导入一下，Reimport，发现还是红色波浪线，报错
clean一下：

文章图片


文章图片

发现BUILD SUCCESS，没问题，说明没有影响，可能是其它的一些jar包没有导进来，不过没有影响的。
好了，有了上面的操作，看源码就很方便了，直接搜索就可以看了。
比如，想看Hadoop的入口类：FileSystem的源码

文章图片

点击一下，就可以看到源码了：

文章图片

词频统计案例开发及上传jar包到服务器并准备测试数据 建个包：

文章图片

建个scala对象

文章图片


文章图片

开发代码：
一开始的三步骤：首先构建一个sparkConf，再构建一个sc，然后关掉sc

package com.ruozedata.spark.com.ruozedata.spark.core import org.apache.spark.{SparkConf, SparkContext}object WordCountApp { def main(args: Array[String]): Unit = { val sparkConf = new SparkConf() val sc = new SparkContext(sparkConf) //业务逻辑 sc.stop() } }

开发业务逻辑代码：wordcount代码

package com.ruozedata.spark.com.ruozedata.spark.coreimport org.apache.spark.{SparkConf, SparkContext}object WordCountApp { def main(args: Array[String]): Unit = { val sparkConf = new SparkConf() val sc = new SparkContext(sparkConf) //args(0)表示传进来的第一个参数 val textFile = sc.textFile(args(0)) val wc = textFile.flatMap(line => line.split("\t")).map((_,1)).reduceByKey(_+_) //打印出来 wc.collect().foreach(println)sc.stop() } }

开发完成，代码打包

文章图片

等一会，就打包成功了：

文章图片

jjar包在这个目录：spark-train\target\spark-train-1.0.jar
然后把jar上传到虚拟机或者服务器上，

[hadoop@hadoop001 lib]$ pwd /home/hadoop/lib [hadoop@hadoop001 lib]$ ls spark-train-1.0.jar

wordcount数据准备：

[hadoop@hadoop001 data]$ hdfs dfs -text /data/wordcount.txt 19/06/06 12:06:00 INFO lzo.GPLNativeCodeLoader: Loaded native gpl library from the embedded binaries 19/06/06 12:06:00 INFO lzo.LzoCodec: Successfully loaded & initialized native-lzo library [hadoop-lzo rev f1deea9a313f4017dd5323cb8bbb3732c1aaccc5] worldworldhello Chinahello peopleperson love

jar包提交Spark应用程序运行 如何提交？看官网：

文章图片

如何使用spark-submit，看帮助，spark-submit --help 看一下。

文章图片

–class的获得，如下：

文章图片

得到：–class com.ruozedata.spark.com.ruozedata.spark.core.WordCountApp
–master 为loacl[2]
application-jar的位置：/home/hadoop/lib/spark-train-1.0.jar
[application-arguments] 参数：hdfs://hadoop001:9000/data/wordcount.txt
汇总一下，如下：

./spark-submit \ --class com.ruozedata.spark.com.ruozedata.spark.core.WordCountApp \ --master local[2] \ /home/hadoop/lib/spark-train-1.0.jar \ hdfs://hadoop001:9000/data/wordcount.txt

运行结果：

文章图片

词频统计案例迭代之输出结果到HDFS 加一个 wc.saveAsTextFile(args(1))：

package com.ruozedata.spark.com.ruozedata.spark.coreimport org.apache.spark.{SparkConf, SparkContext}object WordCountApp { def main(args: Array[String]): Unit = { val sparkConf = new SparkConf() val sc = new SparkContext(sparkConf)val textFile = sc.textFile(args(0)) val wc = textFile.flatMap(line => line.split("\t")).map((_,1)).reduceByKey(_+_)//wc.collect().foreach(println) wc.saveAsTextFile(args(1))sc.stop() } }

再打包上传，替换上面的jar包，再运行代码：

./spark-submit \ --class com.ruozedata.spark.com.ruozedata.spark.core.WordCountApp \ --master local[2] \ /home/hadoop/lib/spark-train-1.0.jar \ hdfs://hadoop001:9000/data/wordcount.txt \ hdfs://hadoop001:9000/data/wcoutput/

结果：

文章图片

词频统计案例迭代之处理多个输入文件 多个文件如下：

[hadoop@hadoop001 lib]$ hdfs dfs -ls /data/wordcount/ Found 5 items -rw-r--r--1 hadoop supergroup49 2019-0 /data/wordcount/wordcount1.txt -rw-r--r--1 hadoop supergroup49 2019-0 /data/wordcount/wordcount2.txt -rw-r--r--1 hadoop supergroup49 2019-0 /data/wordcount/wordcount3.txt -rw-r--r--1 hadoop supergroup49 2019-0 /data/wordcount/wordcount4.txt -rw-r--r--1 hadoop supergroup49 2019-0 /data/wordcount/wordcount5.txt

执行代码：

./spark-submit \ --class com.ruozedata.spark.com.ruozedata.spark.core.WordCountApp \ --master local[2] \ /home/hadoop/lib/spark-train-1.0.jar \ hdfs://hadoop001:9000/data/wordcount/ \ hdfs://hadoop001:9000/data/wcoutput2/

执行结果如下：

文章图片

词频统计案例迭代之输入文件规则匹配 文件如下：

[hadoop@hadoop001 lib]$ hdfs dfs -ls /data/wordcount/ Found 6 items -rw-r--r--1 hadoop supergroup49 2019-0 /data/wordcount/test -rw-r--r--1 hadoop supergroup49 2019-0 /data/wordcount/wordcount1.txt -rw-r--r--1 hadoop supergroup49 2019-0 /data/wordcount/wordcount2.txt -rw-r--r--1 hadoop supergroup49 2019-0 /data/wordcount/wordcount3.txt -rw-r--r--1 hadoop supergroup49 2019-0 /data/wordcount/wordcount4.txt -rw-r--r--1 hadoop supergroup49 2019-0 /data/wordcount/wordcount5.txt

执行代码

./spark-submit \ --class com.ruozedata.spark.com.ruozedata.spark.core.WordCountApp \ --master local[2] \ /home/hadoop/lib/spark-train-1.0.jar \ hdfs://hadoop001:9000/data/wordcount/*.txt \//这里是文件规则匹配 hdfs://hadoop001:9000/data/wcoutput3/

结果如下：
还是5个，说明上面的那个test文件并没有作为输入。

文章图片

带排序的词频统计案例开发及运行过程深度剖析 代码如下：

package com.ruozedata.spark.com.ruozedata.spark.coreimport org.apache.spark.{SparkConf, SparkContext}object WordCountApp { def main(args: Array[String]): Unit = { val sparkConf = new SparkConf() val sc = new SparkContext(sparkConf)val textFile = sc.textFile(args(0)) val wc = textFile.flatMap(line => line.split("\t")).map((_,1)).reduceByKey(_+_) val sorted = wc.map(x => (x._2,x._1)).sortByKey(false).map(x => (x._2,x._1)) //这个在上一节已经有详细介绍 //wc.collect().foreach(println) sorted.saveAsTextFile(args(1))sc.stop() } }

打包上传，替换原有jar包，再执行代码：

./spark-submit \ --class com.ruozedata.spark.com.ruozedata.spark.core.WordCountApp \ --master local[2] \ /home/hadoop/lib/spark-train-1.0.jar \ hdfs://hadoop001:9000/data/wordcount/*.txt \ hdfs://hadoop001:9000/data/wcoutput4/

执行结果如下：
降序：

文章图片

求用户访问量的TopN的Hive实现以及Spark Core实现过程分析 先用hive来实现：
创建一张表：

create table page_views( track_time string, url string, session_id string, referer string, ip string, end_user_id string, city_id string ) row format delimited fields terminated by '\t';

加载数据进来：

load data local inpath '/home/hadoop/data/page_views.dat' overwrite into table page_views;

查询结果：

hive (g6)> select end_user_id,count(*) countfrom page_views group by end_user_id order by count desc limit 5; .......省略.... end_user_idcount NULL60871 12362664840 11619144738 12276468034 8525241930 Time taken: 43.963 seconds, Fetched: 5 row(s) hive (g6)>

可以看到每个用户的id以及相应的访问数量。
再用Spark Core实现：求用户访问量的 top5 。
拿到需求，进行分析，然后功能拆解（中文描述出来，详细设计说明书），最后是代码的开发实现。
首先拿到一个文件，里面有很多行，每行好几个字段，用户的id在第六个字段，每个字段之间按照\t键进行分割。
所以需要拿到用户id，拿到了之后，后面就跟wordcount案例一样了，首先每个用户赋值为1，然后根据用户id分组求总次数，就是根据相同的key进行相加，然后反转，降序，最后再反转，取top5输出。
代码实现：

package com.ruozedata.spark.com.ruozedata.spark.core import org.apache.spark.{SparkConf, SparkContext}object PageViewsApp { def main(args: Array[String]): Unit = { val sparkConf = new SparkConf() val sc = new SparkContext(sparkConf) val pageViews = sc.textFile(args(0))//获取用户id //按照tab键分割，分割之后拿到第六个字段(从0开始)， //然后每个元素赋值1，形成tuple元组 val userids = pageViews.map(x => (x.split("\t")(5),1)) //和wordcount一样 val result = userids.reduceByKey(_+_).map(x => (x._2,x._1)).sortByKey(false).map(x => (x._2,x._1)).take(5).foreach(println)sc.stop() } }

然后打包上传，运行代码：

spark-submit \ --class com.ruozedata.spark.com.ruozedata.spark.core.PageViewsApp \ --master local[2] \ /home/hadoop/lib/spark-train-1.0.jar \ /home/hadoop/data/page_views.dat

运行结果，和上面hive运行的结果一样：

文章图片

就像上面一样，工作中很多场景都可以看到wordcount的影子。
求平均年龄的Spark Core实现 数据格式为：ID + " " + 年龄，如下：

[hadoop@hadoop001 data]$ cat agedata.txt 1 12 2 34 3 54 4 3 5 33 6 23 7 12 8 54 9 45 10 28

分析步骤：
①取出年龄
②求出人数
③年龄相加/人数
开发代码：

package com.ruozedata.spark.com.ruozedata.spark.coreimport org.apache.spark.{SparkConf, SparkContext}object AvgAgeCalculatorApp { def main(args: Array[String]): Unit = { val sparkConf = new SparkConf() val sc = new SparkContext(sparkConf) val dataFile = sc.textFile(args(0))val ageData = https://www.it610.com/article/dataFile.map(x => x.split(" ")(1)) val totalAge = ageData.map(x => x.toInt).reduce(_+_) val count = ageData.count() val avgAge = totalAge/count println(avgAge)sc.stop() } }

打包上传，执行命令：

spark-submit \ --class com.ruozedata.spark.com.ruozedata.spark.core.AvgAgeCalculatorApp \ --master local[2] \ /home/hadoop/lib/spark-train-1.0.jar \ /home/hadoop/data/agedata.txt

即可得到结果。
求男女人数以及最高和最低身高 需求：
1）统计男女人数
2）男性中最高身高和最低身高
3）女性中最高身高和最低身高
数据格式为：ID + " " + 性别 + " " + 身高
如下：

[hadoop@hadoop001 data]$ cat people_info.txt 1 M 178 2 M 156 3 F 165 4 M 178 5 F 154 6 M 167 7 M 189 8 F 210 9 F 209 10 F 190 11 M 176 12 M 165 13 F 159 14 M 155 15 M 164

代码如下：

package com.ruozedata.spark.com.ruozedata.spark.coreimport org.apache.spark.{SparkConf, SparkContext}object peopleApp { def main(args: Array[String]): Unit = { val sparkConf = new SparkConf() val sc = new SparkContext(sparkConf) val peopleData = https://www.it610.com/article/sc.textFile(args(0))//第一步分别取出 男性+身高以及女性+身高 val mRdd = peopleData.map(x =>x.split(" ")).map(line => (line(1),line(2))).filter(_._1 == "M") val fRdd = peopleData.map(x =>x.split(" ")).map(line => (line(1),line(2))).filter(_._1 == "F")//求男女人数 val mCount = mRdd.count() val fCount = fRdd.count()//求男性中最高身高和最低身高 val mMaxHeight =mRdd.map(x => (x._2).toInt)max() //也可以这样求：peopleData.map(line =>line.split(" ")).filter(line => line(1)=="M").sortBy(x => x(2),false).take(1) //peopleData.map(line =>line.split(" ")).filter(line => line(1)=="M").map(x =>x(2).toInt).max val mMinHeight = mRdd.map(x => (x._2).toInt).min()//求女性中最高身高和最低身高 val fMaxHeight = fRdd.map(x => (x._2).toInt).max() //也可以这样求：peopleData.map(line =>line.split(" ")).filter(line => line(1)=="F").sortBy(x => x(2),false).take(1) //peopleData.map(line =>line.split(" ")).filter(line => line(1)=="F").map(x =>x(2).toInt).max val fMinHeight = fRdd.map(x => (x._2).toInt).min()println("男性人数:" + mCount + "" + "女性人数:" + fCount) println("男性中最高身高为:" + mMaxHeight + "男性中最低身高为:" + mMinHeight) println("女性中最高身高为:" + fMaxHeight + "女性中最低身高为:" + fMinHeight)sc.stop() } }

【IDEA开发Spark应用程序】下图只是部分测试结果，可参考：

文章图片

推荐阅读

• 

  Docker容器下快速搭建Hyperf框架在Windows系统的本地开发环境
• 

  u盘启动快捷键u盘启动快捷键
• 

  吃了止痛药能打新冠疫苗吗
• 

  FGO魔伊复刻流程攻略 魔法少女纪行复刻活动流程推荐
• 

  随性写的
• 

  高三学生经商有前途吗？
• 

  让你的打印机重获新生 益普生打印机废墨清零软件下载
• 

  3000元价位手机怎么选好 3000元价位手机怎么选
• 

  安乐什么意思 安乐是什么意思
• 

  王者荣耀狂铁所有皮肤 王者荣耀狂铁战令皮肤怎么样
• 

  代理服务器出了什么问题？ 代理服务器不能怎么回事
• 

  纵向沟通与横向沟通有何不同 横向沟通与纵向沟通各有哪些优缺点
• 

  吃南瓜子能减肥吗
• 

  圆的面积和底面积一样吗? 圆的面积和底面积一样吗
• 

  手机cpu怎么设置最佳性能
• 

  宫腔镜|不知道宫腔镜有何优势？给你列出这3点，可别小瞧了
• 

  急脾气的妈妈怎么对待慢性子的孩子
• 

  粉扑拿回来要洗吗 粉扑用完怎么放置
• 

  的哥马上要交班，却闯祸又丢工作
• 

  mysql查询前三名 mysql查询第三到第五

• 深入理解Go之generate
• 标签、语法规范、内联框架、超链接、CSS的编写位置、CSS语法、开发工具、块和内联、常用选择器、后代元素选择器、伪类、伪元素。
• Spark|Spark 数据倾斜及其解决方案
• IDEA|IDEA 创建工程
• 15、IDEA学习系列之其他设置（生成javadoc、缓存和索引的清理等）
• 我的软件测试开发工程师书单
• echart|echart 双轴图开发
• NPDP拆书（三）（新产品开发战略（经营与创新战略））
• 芯灵思SinlinxA33开发板Linux内核定时器编程
• 常用git命令总结