Spark项目实战:购物网站评价标签生成(非常详细的Spark算子操作)

实战概览

一、项目简介
- 1. 需求
- 2. 内容
二、项目的开发环境
三、项目代码编写
- 1. 项目搭建
- 2. 分析原始的数据
- 3. 编写JSON解析类
- 4. 编写数据处理类
- 5. 将项目打包成jar提交到集群
四、运行结果
五、总结

一、项目简介

1. 需求

如今我们已经离不开没有电商的生活，在上面购物后会做出相应的评价，电商会为那些评价打上标签，比如这样：

通过评价抽取标签有多种方法，有人工方法抽取，机器抽取等等，现标签已经由机器抽取生成，要对其聚合以标签出现的频次作为参考依据，作为该商家的评论标签。

2. 内容

分析原始的用户评价数据，分析并清洗数据，得到用户的评价标签，并选出每个商家评价最多的10评价标签价作为该商家的展示标签。本次会使用Spark进行数据的处理，体验Spark算子处理数据的便捷与魅力。

二、项目的开发环境

1. Spark版本：spark-2.2.0-bin-2.6.0-cdh5.7.0
2. JDK版本1.8，Scala版本2.11
3. 开发工具为IDEA2018

三、项目代码编写

1. 项目搭建

(1). 构建Scala工程，添加Maven支持,创建包目录：

(2). 添加完整依赖

<properties><scala.version>2.11.8</scala.version><spark.version>2.2.0</spark.version>
</properties><repositories><repository><id>cloudera</id><url>https://repository.cloudera.com/artifactory/cloudera-repos</url></repository></repositories><dependencies><dependency><groupId>org.scala-lang</groupId><artifactId>scala-library</artifactId><version>${scala.version}</version></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-core_2.11</artifactId><version>${spark.version}</version></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>1.2.41</version></dependency></dependencies>
<build><pluginManagement><plugins><plugin><artifactId>maven-clean-plugin</artifactId><version>3.0.0</version></plugin><!-- see http://maven.apache.org/ref/current/maven-core/default-bindings.html#Plugin_bindings_for_jar_packaging --><plugin><artifactId>maven-resources-plugin</artifactId><version>3.0.2</version></plugin><plugin><artifactId>maven-compiler-plugin</artifactId><version>3.7.0</version></plugin><plugin><artifactId>maven-surefire-plugin</artifactId><version>2.20.1</version></plugin><plugin><artifactId>maven-jar-plugin</artifactId><version>3.0.2</version></plugin><plugin><artifactId>maven-install-plugin</artifactId><version>2.5.2</version></plugin><plugin><artifactId>maven-deploy-plugin</artifactId><version>2.8.2</version></plugin></plugins></pluginManagement><plugins><plugin><groupId>org.scala-tools</groupId><artifactId>maven-scala-plugin</artifactId><executions><execution><goals><goal>compile</goal><goal>testCompile</goal></goals></execution></executions><configuration><scalaVersion>${scala.version}</scalaVersion><args><arg>-target:jvm-1.5</arg></args></configuration></plugin><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-eclipse-plugin</artifactId><configuration><downloadSources>true</downloadSources><buildcommands><buildcommand>ch.epfl.lamp.sdt.core.scalabuilder</buildcommand></buildcommands><additionalProjectnatures><projectnature>ch.epfl.lamp.sdt.core.scalanature</projectnature></additionalProjectnatures><classpathContainers><classpathContainer>org.eclipse.jdt.launching.JRE_CONTAINER</classpathContainer><classpathContainer>ch.epfl.lamp.sdt.launching.SCALA_CONTAINER</classpathContainer></classpathContainers></configuration></plugin></plugins></build>

2. 分析原始的数据

本次所用于分析的数据比较多就不好直接放出来了，大家可以取自行下载：购物网站评价标签生成原始数据，截取其中一条数据：

其中77287793 是商家编号，由制表符隔开的是一个{…}，这是JSON数据，其中Key"extInfoList"为一个JSON数组，其中的一个组员"Value"包含我们想要得到的评价，是一个字符串数组。所以我们要做的首先对数据进行清洗，然后解析JSON数据，得到评价字符串，解析使用Java脚本，然后使用Spark算子对其进行处理。

3. 编写JSON解析类

(1). 在Java目录的util包下创建 ReviewTags 类，用于解析JSON数据：

(2). ReviewTags 类完整代码如下：

/*** 取出标签的评语,拼装成以","分割的字符串*/
public class ReviewTags {public static String extractTags(String jsonStr){//将字符串封装为JSON对象JSONObject object = JSON.parseObject(jsonStr);//判断JSON对象是否为空或者是否包含Key"extInfoList"if(object == null || !object.containsKey("extInfoList")){return "";}//extInfoList对应的Value是一个JSON数组//"extInfoList":[{"title":"contentTags","values":["干净卫生","服务热情"],"desc":"","defineType":0},{"title":"tagIds","values":["852","22"],"desc":"","defineType":0}]JSONArray array = object.getJSONArray("extInfoList");if(array == null) {return "";}StringBuilder sb = new StringBuilder();for (int i = 0; i < array.size(); i++) {//取出JSON数组每一个元素，为JSON对象JSONObject obj = array.getJSONObject(i);//验证JSON数组的元素是否为目标元素if (obj != null && obj.containsKey("title") && obj.getString("title").equals("contentTags") && obj.containsKey("values")) {//取出评价标签的内容，是一个字符串数组JSONArray arr = obj.getJSONArray("values");if(arr == null){continue;}boolean begin = true;//拼接评价,生成："干净卫生,服务热情"for (int j = 0; j < arr.size(); j++) {//使得第一个字符串前无","if (begin) {begin = false;} else {sb.append(",");}sb.append(arr.getString(j));}}}//返回字符串return sb.toString();}
}

4. 编写数据处理类

(1). 在Scala目录的app包下创建 Scala的Object类 TagGenerator ，用于处理数据：

(2). TagGenerator 类完整代码如下(有详细注释)：

/***数据处理类*/
object TagGenerator {def main(args: Array[String])= {//检验是否已经给定合法的参数if(args.length != 2){println("Yo should input <InputPath> <OutputPath>")System.exit(1)}//将args里面的参数取出来val inputPath = args(0)val outputPath = args(1)//创建parkConfval sparkConf = new SparkConf().setAppName("TagGenerator")//创建SparkContextval sc = new SparkContext(sparkConf)//从给定的HDFS路径中取出数据val poi_tags = sc.textFile(inputPath)//使用制表符分割原始数据,得到一个字符串数组，长度为2val poi_taglist = poi_tags.map(e=>e.split("\t")).//过滤掉非法数据，如果数组长度不为2则说明数据有缺失，要过滤filter(e=>e.length == 2).//e为数组，e(0)为商家编号，然后做一个映射，将评价字符串与与商家编号做映射//得到这样的元组：(77287793,"音响效果好,体验好,价格实惠")map(e => e(0) -> ReviewTags.extractTags(e(1))).//e._2.length是指元组第二个元素的长度，即"音响效果好,体验好,价格实惠"，若小于0则过滤filter(e => e._2.length > 0).//将商家编号取出来，并将商家的评价字符串以","分割，得到评价数组，并以商家编号映射该数组//得到这样的元组：(77287793,("音响效果好","体验好","价格实惠"))map(e => e._1 -> e._2.split(",")).//将商家编号分别映射数组的每个元素,得到以下元组：//(77287793,"音响效果好"),(77287793,"体验好") ...flatMapValues(e => e).//将元组再次映射为新的元组，原来的元组为新元组的Key，Value为1：//((77287793,"音响效果好"),1),((77287793,"体验好"),1),...map(e => (e._1,e._2) -> 1).//对Key进行聚合，得到结果为：//((77287793,"音响效果好"),180),((77287793,"体验好"),95),...reduceByKey(_+_).//e._1._1代表取出元组的第一个组员的第一个组员，即为商家编号，来映射一个List//List的元素为元组，第一个组员为e._1._2，即为评价"音响效果好",e._2为评价数，为180，从而得到以下结果：//(77287793,[("音响效果好",180),("体验好",95),...])map(e => e._1._1 -> List((e._1._2,e._2))).//这一步非常之巧妙，对Key进行聚合，但是Key为列表List，所以采用:::,表示List的追加，从而把商家的所有评价都聚集在了一起//(77287793,[("音响效果好",180),("体验好",95),...])reduceByKey(_:::_).//这一步工作比较多，首先将商家编号取出，然后对List基于评价数(_._2)进行升序排序,//然后对排序结果反转(reverse),取前10条(take(10))，即评价人数最多的10条，然后将那10个元组取出//变为评价:评论数的字符串列表：List("音响效果好:180","体验好,95",...),最后对List使用mkString方法，生产//以逗号隔开的字符串,并以商家号映射：(77287793,"音响效果好:180,体验好,95,...")map(e => e._1 -> e._2.sortBy(_._2).reverse.take(10).map(a => a._1+":"+a._2.toString).mkString(","))//将结果的元组取出来，并将其转变为以制表符分割的字符串poi_taglist.map(e=>e._1+"\t"+e._2).//将统计结果存入给定的HDFS路径中以便分析和运用saveAsTextFile(outputPath)}
}

5. 将项目打包成jar提交到集群

这里有个问题要注意，因为我们把项目打包提交到Spark集群上运行，但是Spark集群是没有JSON类的jar包的，所以我们需要打依赖jar包，即把项目所依赖的jar包也包含进去，这需要额外的Maven插件来完成，但是Spark集群本身包含Spark的jar包，如果再把Saprk的jar打包进去是没必要且会使得jar变得很大，所以我们要使用Maven提供的一个域标签隔离Spark，使得Spark的jar包只在本地有效，使其在打包的时候不被打进去。
(1). 在<build>标签的的<plugins>下再添加以下插件：

 <plugin><artifactId>maven-assembly-plugin</artifactId><executions><execution><phase>package</phase><goals><goal>single</goal></goals></execution></executions><configuration><descriptorRefs><descriptorRef>jar-with-dependencies</descriptorRef></descriptorRefs></configuration>
</plugin>

(2). 修改pom.xml文件的Spark的依赖配置，使其不能将其jar包也打进我们的项目jar包中：

(3). 在项目本地目录下的target文件下可以看到两个jar，选择依赖jar上传至集群：

四、运行结果

(1). 将下载好的原始数据上传至HDFS：
hdfs dfs -put temptags.txt /data
(2). 执行命令运行程序
如果你部署了Spark集群，可以执行如下命令:spark-submit --master spark://hadoop00:7077 --jars temptag-1.0-jar-with-dependencies.jar --class com.temptag.app.TagGenerator hdfs://hadoop00:/data/temptags.txt hdfs://hadoop00:/result/out
后面的两个路径分别是原始数据的路径，计算结果的路径,顺序不可以改变。
如果你没有部署Spark集群，但是安装了Saprk，可以执行如下命令:spark-submit --master local[2] --jars temptag-1.0-jar-with-dependencies.jar --class com.temptag.app.TagGenerator hdfs://hadoop00:/data/temptags.txt hdfs://hadoop00:/result/out
(3). 执行结果：

五、总结

本次的实战中，我们首先在HDFS上读取原始数据，使用Java脚本解析JSON数据，然后使用Spark算子对数据进行清洗，映射，聚合，排序等工作，最后将计算结果存储在HDFS中。从中我们发现通过使用Spark算子的组合，使得繁琐复杂的数据处理工作变得相当的简洁，这也是Spark的优势之一。我是人间，乐于结交共同学习的朋友，欢迎大家一起交流，共同进步！