《企业实训-推荐系统实践》实验报告

1、 企业实训-推荐系统实践实 验 报 告实 验 名 称： 智慧交通出行之Spark SQL业务分析 学 生 姓 名： 罗霄 班 级 软件183 学 号 学 院（系） 阿里云大数据学院 指导教师： 雷晓孟 实 验 时 间： 2021 年 10 月 13 日 一、实验准备1.1实验概述 随着人们对出行的需求日益增加，出行的安全问题，出行的便捷问题等问题日益突出，特别是安全出行是我们每个人都迫切需要的，为了增加出行的便捷，提高出行的安全，对我们乘车的细节以及发生点我们迫切的需要及时知道，为此特地通过大数据的手段来处理我们海量的出行数据，做到订单的实时监控，乘车轨迹的的细节回放，虚拟

2、车站的科学制定，出行迁途的细节过程，订单报表的大屏展示，以及用户乘车行为统计，用户画像等功能，实现用户的出行统计，制定用户的的“杀熟”策略等。1.2实验知识点·学会Linux命令基础操作·学会Spark SQL框架数据分析1.3实验环境·CentOS 7操作系统·Jdk-1.8 环境·Hadoop-2.7环境框架·Zookeeper-3.4环境框架·Hbase-1.3环境框架·Spark-2.1环境框架1.4预备知识点·Linux基础操作命令·Hadoop框架的基础知识·My

3、SQL框架的基础知识·Spark 框架的基础知识二、实验原理大数据智慧交通出行项目案例主要介绍流程，通过日志数据和业务数据，我们使用Flume框架作为数据的采集，数据采集到之后使用Kafka框架作为数据的消费，然后把存储数据到Hbase数据库中。同时把MySQL业务数据库中数据，通过Spark Sql对数据业务进行分析。三、实验步骤步骤1：Spark WordCount例操作1.1点击又上角“切换终端”，通过终端命令窗口编写实验代码1.2 hadoop启动   Hadoop是一个由Apache基金会所开发的分布式系统基础架构。主要解决，海量数据

4、的存储和海量数据的分析计算问题。Hadoop的优势：1）高可靠性：Hadoop底层维护多个数据副本，所以即使Hadoop某个计算元素或存储出现故障，也不会导致数据的丢失。2）高扩展性：在集群间分配任务数据，可方便的扩展数以千计的节点。3）高效性：在MapReduce的思想下，Hadoop是并行工作的，以加快任务处理速度。4）高容错性：能够自动将失败的任务重新分配。提示：由于hive操作的数据是 HDFS 中的数据，所以必须先启动 hadoop。source /etc/profile       

5、              #刷新系统环境变量                                   &#

6、160;  #hadoop启动命令jps                                              

7、0;        #查看启动注意： 启动hadoop时，会把hadoop中各个组件启动，此时需要输入多次密码1.3 启动yarn命令   YARN 是Hadoop2.0版本新引入的资源管理系统，直接从MR1演化而来，主要由ResourceManager、NodeManager、ApplicationMaster、和Container组成。          

8、0;    #启动yarn命令jps                                    #查看启动程序如图所示1.4启动MySQL数据库#开启mysql服务service 

9、;mysql start1.5鼠标点击右键，选择 “Open Terminal Here”如图所示：1.6启动idea编辑器   IDEA 全称 IntelliJ IDEA，是java编程语言开发的集成环境。IntelliJ在业界被公认为最好的java开发工具，尤其在智能代码助手、代码自动提示、重构、JavaEE支持、各类版本工具(git、svn等)、JUnit、CVS整合、代码分析、 创新的GUI设计等方面的功能可以说是超常的。 #更新环境source /etc/profile  #进入到idea文件的bin目录

10、下cd /idea/bin #启动命令 ./1.7选择 Evaluate for free1.8这里选择 Evaluate for free 选项，然后点击 Evaluate 按钮。1.9创建一个Java项目,打开 IntelliJ IDEA 编辑器软件，点击 Create New Project提示： 第一步 通过使用 IntelliJ IDEA 编辑器，创建Java项目。第二步通过创建好的Java项目，在里面创建Java类。如图所示：1.10选择 创建

11、Scala项目，然后选中 IDEA ，完成后点击 Next下一步       Scala 是 Scalable Language 的简写，是一门多范式的编程语言.Scala是一种纯面向对象的语言，每个值都是对象。对象的数据类型以及行为由类和特质描述。类抽象机制的扩展有两种途径：一种途径是子类继承，另一种途径是灵活的混入机制。这两种途径能避免多重继承的种种问题。Scala也是一种函数式语言，其函数也能当成值来使用。Scala提供了轻量级的语法用以定义匿名函数，支持高阶

12、函数，允许嵌套多层函数，并支持柯里化。Scala的case class及其内置的模式匹配相当于函数式编程语言中常用的代数类型。更进一步，程序员可以利用Scala的模式匹配，编写类似正则表达式的代码处理XML数据。Scala具备类型系统，通过编译时检查，保证代码的安全性和一致性。1.11项目名称 Project name 填写为 “ScalaProject”,Scala Sdk 我们通过 Cteate. 创建选择，点击 create.，进入选项1.12 编辑器会默认检索出系统环境变量的Scala，此时点击 

13、“OK”  1.12最后展示出 Scala SDK 环境，然后点击 Finish 完成总结：【1】填写项目名称，【2】选择JDK1.8版本(默认已选择好)，【3】选择scala-sdk-2.11.12版本， Scala SDK版本默认第一次不会选择好，需要自己点击 Create 创建选中scala-sdk-2.11.11版本1.13欢迎提示页面，直接点击关闭欢迎页面。 11.如下图所示，新建包“sparkpro”下新建“”对象类12.编写WordCount 代码 packag

14、e sparkpro import e.spark.SparkContext, SparkConf  object WordCount    def main(args: ArrayString)       val conf = new SparkConf().setAppName("WordCount").setMaster("local*") 

15、;    /创建SparkContext 对象     val sc = new SparkContext(conf)     / 读取本地文件     val lines = sc.textFile("/opt/Datas/input/wc.txt")       / 拆分单词，并过滤掉空格，当然还可以继续进行过滤，如去掉标点

16、符号     val words = ap(_.split(" ")  / 在单词拆分的基础上对每个单词实例计数为1, 也就是 word => (word, 1)     val pairs = (word => (word, 1)    / 在每个单词实例计数为1的基础之上统计每个单词在文件中出现的总次数, 即ke

17、y相同的value相加     val wordscount = eByKey(_ + _)    / 打印结果     ct.foreach(println)       / 释放资源     sc.stop()     部分截图运行代码结果小结：此时统计出单词的个数，格式为（单词名，次数） 步骤2：Spark出租车数据分析&

18、#160;  Apache Spark 是专为大规模数据处理而设计的快速通用的计算引擎。Spark是UC Berkeley AMP lab 所开源的类Hadoop MapReduce的通用并行框架，Spark，拥有Hadoop MapReduce所具有的优点；但不同于MapReduce的是Job中间输出结果可以保存在内存中，从而不再需要读写HDFS，因此Spark能更好地适用于数据挖掘与机器学习等需要迭代的MapReduce的算法。Spark 是一种与 Hadoop 相似的开源集群计算环境，但是两者之间还存在一些不同之处，这

19、些有用的不同之处使 Spark 在某些工作负载方面表现得更加优越，换句话说，Spark 启用了内存分布数据集，除了能够提供交互式查询外，它还可以优化迭代工作负载。2.1 在idea编辑器中新建 “Tax1” 对象类编写Tax1数据分析代码使用训练好的模型对测试数据进行预测，然后对结果以（类别，小时时间 ）进行count统计，结果为每个类别每个小时的总次数。 package Java  import e.s import e.rAssembler import e.R

20、ow, SparkSession import e.functions.desc, substring import e.types._  object Tax1    def main(arg:ArrayString):Unit=     val spark = er().appName("Tax1").master("local*").getOrCreate()&

21、#160;    /为读取的数据创建schema     val taxiSchema = StructType(Array(       StructField("id",IntegerType,true),       StructField("tid",DoubleType,true),       StructField("lat"

22、;,DoubleType,true),       StructField("time",StringType,true)    )     val path = "D:OS"     val data = .schema(taxiSchema).csv(path)     /将tid和lat转换成训练使用的Vector类型   

23、;  val assembler = new VectorAssembler()     val tid_lat = Array("tid","lat")          val vecDF = putCols(tid_lat).setOutputCol("features").transform(data)   &

24、#160;  val splitRait = Array(0.8, 0.2)     val Array(train, test) = mSplit(splitRait)      /按照8：2的比例随即分割数据，分别用于训练和测试     /val splitRait = Array(0.8, 0.2)     /val Array(train, test) = mSplit

25、(splitRait)     /建立Kmeans，设置类别数为10     val kmeans = new KMeans().setK(10).setFeaturesCol("features").setPredictionCol("prediction")     /模型训练     val model = (train)      /使用模型预测 测试数据     val testResul