《通信数据分析与实战》课件-第二章 Spark 基础

通信数据分析与实战Spark基础第二章第1节2知道Spark的作用了解Spark的生态圈熟悉Spark的特点学习目标TARGETSpark于2009年诞生于美国加州大学伯克利分校的AMP实验室，它是一个可应用于大规模数据处理的统一分析引擎。Spark不仅计算速度快，而且内置了丰富的API，使得我们能够更加容易编写程序。

Spark在2013年加入Apache孵化器项目，之后获得迅猛的发展，并于2014年正式成为Apache软件基金会的顶级项目。Spark生态系统已经发展成为一个可应用于大规模数据处理的统一分析引擎，它是基于内存计算的大数据并行计算框架，适用于各种各样的分布式平台的系统。Spark的概述Spark生态圈中包含了SparkSQL、SparkStreaming、GraphX、MLlib等组件。Spark的生态圈SparkCore和SparkSQLSpark的生态圈SparkCoreSparkSQLSpark核心组件，实现了Spark的基本功能，包含任务调度、内存管理、错误恢复、与存储系统交互等模块。SparkCore中还包含对弹性分布式数据集的API定义。用来操作结构化数据的核心组件，通过SparkSQL可直接查询Hive、HBase等多种外部数据源中的数据。SparkSQL的重要特点是能够统一处理关系表和RDD。SparkStreaming和MLlibSpark的生态圈MLlibSpark提供的流式计算框架，支持高吞吐量、可容错处理的实时流式数据处理，其核心原理是将流数据分解成一系列短小的批处理作业。SparkStreamingSpark提供的关于机器学习功能的算法程序库，包括分类、回归、聚类、协同过滤算法等，还提供了模型评估、数据导入等额外的功能。GraphX和调度器Spark的生态圈

GraphX独立调度器、Yarn、MesosSpark提供的分布式图处理框架，拥有对图计算和图挖掘算法的API接口及丰富的功能和运算符，便于对分布式图处理的需求，能在海量数据上运行复杂的图算法。集群管理器，负责Spark框架高效地在一个到数千个节点之间进行伸缩计算的资源管理。Spark的特点Spark计算框架在处理数据时，所有的中间数据都保存在内存中，从而减少磁盘读写操作，提高框架计算效率。同时Spark还兼容HDFS、Hive，可以很好地与Hadoop系统融合，从而弥补MapReduce高延迟的性能缺点。所以说，Spark是一个更加快速、高效的大数据计算平台。spark速度快易用性通用性兼容性Spark的应用场景数据科学数据工程师可以利用Spark进行数据分析与建模，由于Spark具有良好的易用性，数据工程师只需要具备一定的SQL语言基础、统计学、机器学习等方面的经验，以及使用Python、Matlab或者R语言的基础编程能力，就可以使用Spark进行上述工作。数据处理大数据工程师将Spark技术应用于广告、报表、推荐系统等业务中，在广告业务中，利用Spark系统进行应用分析、效果分析、定向优化等业务，在推荐系统业务中，利用Spark内置机器学习算法训练模型数据，进行个性化推荐及热点点击分析等业务。10小结知道Spark的作用了解Spark的生态圈熟悉Spark的特点通信数据分析与实战Spark基础第二章第2节12知道Spark的部署方式掌握Spark的安装过程熟悉Spark的集群部署流程学习目标TARGETSpark的搭建的环境介绍由于Spark仅仅是一种计算框架，不负责数据的存储和管理，因此，通常都会将Spark和Hadoop进行统一部署，由Hadoop中的HDFS、HBase等组件负责数据的存储管理，Spark负责数据计算。

所以在搭建Spark环境之前,务必安装好Hadoop的集群环境Spark的部署方式Standalone模式Standalone模式被称为集群单机模式。该模式下，Spark集群架构为主从模式，即一台Master节点与多台Slave节点，Slave节点启动的进程名称为Worker，存在单点故障的问题。Mesos模式被称为SparkonMesos模式。Mesos是一款资源调度管理系统，为Spark提供服务，由于Spark与Mesos存在密切的关系，因此在设计Spark框架时充分考虑到对Mesos的集成。Yarn模式被称为SparkonYarn模式，即把Spark作为一个客户端，将作业提交给Yarn服务。由于在生产环境中，很多时候都要与Hadoop使用同一个集群，因此采用Yarn来管理资源调度，可以提高资源利用率。Spark集群安装部署我们将在Standalone模式下，进行Spark集群的安装部署。规划的Spark集群包含一台Master节点和两台Slave节点。其中，主机名hadoop01是Master节点，hadoop02和hadoop03是Slave节点。Spark集群安装部署1、下载Spark安装包Spark集群安装部署2、解压Spark安装包

首先将下载的spark-2.3.2-bin-hadoop2.7.tgz安装包上传到主节点hadoop01的/export/software目录下，然后解压到/export/servers/目录，解压命令如下。tar-zxvfspark-2.3.2-bin-hadoop2.7.tgz-C/export/servers/Spark集群安装部署3、修改配置文件1（1）将spark-env.sh.template配置模板文件复制一份并命名为spark-env.sh

（2）修改spark-env.sh文件，在该文件添加以下内容：#配置java环境变量exportJAVA_HOME=/export/servers/jdk#指定Master的IPexportSPARK_MASTER_HOST=hadoop01#指定Master的端口

exportSPARK_MASTER_PORT=7077Spark集群安装部署3、修改配置文件2（3）复制slaves.template文件，并重命名为slaves

（4）修改spark-env.sh文件，在该文件添加以下内容：通过“vislaves”命令编辑slaves配置文件，主要是指定Spark集群中的从节点IP，由于在hosts文件中已经配置了IP和主机名的映射关系，因此直接使用主机名代替IP，添加内容如下。hadoop02hadoop03Spark集群安装部署4、分发文件修改完成配置文件后，将spark目录分发至hadoop02和hadoop03节点scp-r/export/servers/spark/hadoop02:/export/servers/scp-r/export/servers/spark/hadoop03:/export/servers/Spark集群安装部署5、启动Spark集群sbin/start-all.sh集群进程Spark集群安装部署5、启动Spark集群Spark集群管理页面SparkHA集群部署SparkStandalone集群是主从架构的集群模式，由于存在单点故障问题，解决这个问题需要用到Zookeeper服务，其基本原理是将Standalone集群连接到同一个Zookeeper实例并启动多个Master节点，利用Zookeeper提供的选举和状态保存功能，可以使一台Master节点被选举，另一台Master节点处于Standby状态。当活跃的Master发生故障时，Standby状态的Master就会被激活，恢复集群调度，整个恢复的过程可能需要1-2分钟。Zookeeper集群安装1.Zookeeper安装包的下载安装由于Zookeeper集群运行需要Java环境支持，所以要提前安装JDK（对于jdk的下载安装这里不作赘述）。Zookeeper安装包的下载安装，具体步骤如下：1下载Zookeeper安装包。Zookeeper的下载地址为：/apache/zookeeper/zookeeper-3.4.10/。上传Zookeeper安装包。将下载完毕的Zookeeper安装包上传至Linux系统/export/software/目录下。23解压安装Zookeeper安装包。Zookeeper集群安装2.Zookeeper的相关配置修改Zookeeper的配置文件。先将zoo_sample.cfg配置文件重命名为zoo.cfg，然后指定dataDir目录、配置服务器编号与主机名映射关系、设置与主机连接的心跳端口和选举端口。步骤1：#进入Zookeeper解压目录下的conf目录,复制配置文件zoo_sample.cfg并重命名zoo.cfg

cpzoo_sample.cfgzoo.cfg

#使用vi编辑器打开zoo.cfgvizoo.cfg

#设置数据文件目录+数据持久化路径dataDir=/export/data/zookeeper/zkdata#设置对应的主机名选举端口号和通信端口号server.1=hadoop01:2888:3888server.2=hadoop02:2888:3888server.3=hadoop03:2888:3888Zookeeper集群安装2.Zookeeper的相关配置#在指定目录下创建zkdata文件夹mkdir-p/export/data/zookeeper/zkdata

#在zkdata文件夹下创建myid文件cd/export/data/zookeeper/zkdataecho1>myid步骤2：创建myid文件。根据配置文件zoo.cfg设置的dataDir目录，创建zkdata文件夹并创建myid文件，该文件里面的内容就是服务器编号。Zookeeper集群安装2.Zookeeper的相关配置exportZK_HOME=/export/servers/zookeeper-3.4.10exportPATH=$PATH:$JAVA_HOME/bin:$HADOOP_HOME/bin:$HADOOP_HOME/sbin:$ZK_HOME/bin步骤3：配置环境变量。执行命令vi/etc/profile对profile文件进行修改，添加Zookeeper的环境变量Zookeeper集群安装2.Zookeeper的相关配置#将hadoop01上的zookeeper分发到hadoop02hadoop03上scp-r/export/servers/zookeeper-3.4.10/hadoop02:/export/servers/scp-r/export/servers/zookeeper-3.4.10/hadoop03:/export/servers/#将myid文件分发到hadoop02hadoop03上,并且修改myid的文件内容,依次修改成2,3scp-r/export/data/zookeeper/hadoop02:/export/data/scp-r/export/data/zookeeper/hadoop03:/export/data/#在hadoop02上面操作echo2>/export/data/zookeeper/zkdata/myid#在hadoop03上面操作echo3>/export/data/zookeeper/zkdata/myid

#将profile文件分发到hadoop02hadoop03上scp/etc/profilehadoop02:/etc/profilescp/etc/profilehadoop03:/etc/profile分发Zookeeper相关文件至其他服务器。先将Zookeeper安装目录、myid文件以及profile文件分发至hadoop02、hadoop03服务器上，并且修改myid的文件内容，依次对应服务器号进行设置，分别为2、3。步骤4：Zookeeper集群安装2.Zookeeper的相关配置#分别在hadoop01,hadoop02,hadoop03服务器上刷新profile配置文件,使环境变量生效source/etc/profile步骤5：环境变量生效。分别在hadoop01、hadoop02、hadoop03服务器上刷新profile配置文件，使环境变量生效。Zookeeper服务器的启动和关闭Zookeeper的启动在hadoop01、hadoop02、hadoop03服务器上使用zkServer.shstart依次启动Zookeeper服务。执行命令“zkServer.shstatus”查看该节点Zookpeer的角色。步骤1：步骤2：Zookeeper服务器的启动和关闭Zookeeper的关闭在hadoop01、hadoop02、hadoop03服务器上执行zkServer.shstop关闭Zookeeper服务。执行命令“zkServer.shstatus”查看该节点Zookpeer的角色。步骤1：步骤2：SparkHA集群部署1启动Zookeeper集群服务。2在spark-env.sh配置文件中注释Master节点的配置参数，并指定Spark使用Zookeeper管理。3在hadoop01主节点启动Spark集群，在hadoop02节点再次启动Master服务。4关闭hadoop01节点中的Master进程，测试SparkHA集群。SparkHA集群部署#依次在三台节点上启动Zookeeper，命令如下:zkServer.shstart1启动Zookeeper集群服务。SparkHA集群部署2在spark-env.sh配置文件中注释Master节点的配置参数，并指定Spark使用Zookeeper管理。#指定Master的IP#exportSPARK_MASTER_HOST=hadoop01#指定Master的端口exportSPARK_MASTER_PORT=7077exportSPARK_DAEMON_JAVA_OPTS="-Dspark.deploy.recoveryMode=ZOOKEEPER-Dspark.deploy.zookeeper.url=hadoop01:2181,hadoop02:2181,hadoop03:2181-Dspark.deploy.zookeeper.dir=/spark“[spark.deploy.recoveryMode]()：设置Zookeeper去启动备用Master模式。[spark.deploy.zookeeper.url]()：指定ZooKeeper的Server地址。[spark.deploy.zookeeper.dir]()：保存集群元数据信息的文件、目录。SparkHA集群部署2在spark-env.sh配置文件中注释Master节点的配置参数，并指定Spark使用Zookeeper管理。scpspark-env.shhadoop02:/export/servers/spark/confscpspark-env.shhadoop03:/export/servers/spark/conf配置完成后，将spark-env.sh分发至hadoop02、hadoop03节点上，保证配置文件统一，命令如下SparkHA集群部署3在hadoop01主节点启动Spark集群，在hadoop02节点再次启动Master服务。在hadoop01主节点使用一键启动脚本启动，命令如下。/export/servers/spark/sbin/start-all.sh在hadoop02节点再次启动Master服务，命令如下。/export/servers/spark/sbin/start-master.sh启动成功后，通过浏览器访问<http://hadoop02:8080>，查看备用Master节点状态SparkHA集群部署4关闭hadoop01节点中的Master进程，测试SparkHA集群。SparkHA集群启动完毕后，为了演示是否解决了单点故障问题，可以关闭在hadoop01节点中的Master进程，用来模拟在生产环境中hadoop01突然宕机，命令如下所示。/export/servers/spark/sbin/stop-master.sh执行命令后，通过浏览器查看http://hadoop01:8080)发现已经无法通过hadoop01节点访问Spark集群管理界面。大约经过1-2分钟后，刷新http://hadoop02:8080页面，可以发现hadoop02节点中的Status值更改为ALIVE，Spark集群恢复正常，说明SparkHA配置有效解决了单点故障问题。脚本启动Zookeeper集群在集群中启动Zookeeper服务时，需要依次在三台服务器上执行启动命令，然而在实际工作应用中，集群数量并非3台，当遇到数十台甚至更多的服务器时，就不得不编写脚本来启动服务了，编写脚本的语言有多种，这里采用Shell语言开发一键启动Zookeeper服务脚本，使用vi创建start_zk.sh文件#!/bin/shforhostinhadoop01hadoop02hadoop03dossh$host"source/etc/profile;zkServer.shstart"echo"$hostzkisrunning"done39小结知道Spark的部署方式掌握Spark的安装过程熟悉Spark的集群部署流程通信数据分析与实战Spark基础第二章第3节41了解Spark架构的基本概念知道Spark的运行架构熟悉Spark的运行流程学习目标TARGET在学习Spark运行架构与工作原理之前，首先需要了解几个重要的概念和术语基本概述Application（应用）运行main()方法并创建SparkContext的进程。Spark上运行的应用。Application中包含一个驱动器进程和集群上的多个执行器进程。ClusterManager（集群管理器）集群上运行应用程序代码的任意一个节点。用于在集群上申请资源的外部服务（如：独立部署的集群管理器、Mesos或者Yarn）。DriverProgram（驱动器）WorkerNode（工作节点）基本概述Executor（执行器）运行main()方法并创建SparkContext的进程。在集群工作节点上为某个应用启动的工作进程，该进程负责运行计算任务，并为应用程序存储数据。Job（作业）每个Job可划分为更小的Task集合，每组任务被称为

Stage一个并行计算作业，由一组任务组成，并由Spark的行动算子（如：save、collect）触发启动。Stage（阶段）Task（任务）Spark集群运行架构Spark运行架构主要由SparkContext、ClusterManager和Worker组成，其中ClusterManager负责整个集群的统一资源管理，Worker节点中的Executor是应用执行的主要进程，内部含有多个Task线程以及内存空间。Spark集群运行架构

Spark应用在集群上运行时，包括了多个独立的进程，这些进程之间通过驱动程序（DriverProgram）中的SparkContext对象进行协调，SparkContext对象能够与多种集群资源管理器（ClusterManager）通信，一旦与集群资源管理器连接，Spark会为该应用在各个集群节点上申请执行器（Executor），用于执行计算任务和存储数据。每个Spark应用程序都有其对应的多个执行器进程。执行器进程在整个应用程序生命周期内，都保持运行状态，并以多线程方式执行任务。​驱动器程序在整个生命周期内必须监听并接受其对应的各个执行器的连接请求，因此驱动器程序必须能够被所有Worker节点访问到。因为集群上的任务是由驱动器来调度的，所以驱动器应该放在离集群Worker节点比较近的机器上。Spark的运行流程Spark的运行流程当一个Spark应用被提交时，根据提交参数创建Driver进程，Driver进程初始化SparkContext对象，由SparkContext负责和ClusterManager的通信以及资源的申请、任务的分配和监控等。Driver进程向ClusterManager申请资源，ClusterManager接收到Application的注册请求后，会使用自己的资源调度算法，在Spark集群的Worker节点上，通知Worker为应用启动多个Executor。Executor创建后，会向ClusterManager进行资源及状态的反馈，便于ClusterManager对Executor进行状态监控，如果监控到Executor失败，则会立刻重新创建。步骤1步骤2步骤3Spark的运行流程Executor会向SparkContext反向注册申请Task。TaskScheduler将Task发送给Worker进程中的Executor运行并提供应用程序代码。当程序执行完毕后写入数据，Driver向ClusterManager注销申请的资源。步骤4步骤5步骤649小结了解Spark架构的基本概念知道Spark的运行架构熟悉Spark的运行流程通信数据分析与实战Spark基础第二章第4节51知道Spark的小程序的调用知道查看小程序的运行结果学习目标TARGET我们可以通过运行Spark官方示例SparkPi，体验Spark集群提交任务的流程。执行下述代码块中的命令，提交SparkPi程序。第一个Spark程序bin/spark-submit--classorg.apache.spark.examples.SparkPi--masterspark://hadoop01:7077--executor-memory1G--total-executor-cores1examples/jars/spark-examples_2.11-2.3.2.jar10提交Spark作业后，观察Spark集群管理界面，其中“RunningApplications”列表表示当前Spark集群正在计算的作业，执行几秒后，刷新界面，在CompletedApplications表单下，可以看到当前应用执行完毕，返回控制台查看输出信息，出现了“Piisroughly3.140691140691141”，说明Pi值已经被计算完毕第一个Spark程序54小结知道Spark的小程序的调用知道查看小程序的运行结果通信数据分析与实战Spark基础第二章第5节56掌握SparkShell的启动掌握SparkShell和HDFS交互学习目标TARGETSpark

Shell介绍Spark-Shell是一个强大的交互式数据分析工具，初学者可以很好的使用它来学习相关API，用户可以在命令行下使用Scala编写Spark程序，并且每当输入一条语句，Spark-Shell就会立即执行语句并返回结果，这就是我们所说的REPL（Read-Eval-PrintLoop，交互式解释器），Spark-Shell支持Scala和Python，如果需要进入Python语言的交互式执行环境，只需要执行“pyspark”命令即可。Spark

Shell启动在spark/bin目录中，执行下列命令进入Spark-Shell交互环境：--master”表示指定当前连接的Master节点<master-url>用于指定Spark的运行模式bin/spark-shell--master<master-url>Spark

Shell启动<master-url>可取的详细值如下所示。下面通过启动Spark-Shell，并且使用Scala语言开发单词计数的Spark程序，现需要有文本文件words.txt（需要在本地创建文件并上传至指定目录）在HDFS中的/spark/test路径下，且文本内容如下：SparkShell读取HDFS文件

hellohadoophellosparkhello

yuanlongpinghellohuangxuhua#在hadoop01上创建words.txt文件cd/export/dataviwords.txt#启动Hadoop集群start-all.sh#创建文件夹在Hadoop上面hadoopfs–mkdir–p/spark/test#上传本地文件到Hadoop指定路径hadoopfs–put/export/data/words.txt/spark/test/SparkShell读取HDFS文件

1整合Spark和HDFS。2启动Hadoop、Spark服务。3启动Spark–Shell编写程序。4退出Spark-Shell客户端。SparkShell读取HDFS文件

exportHADOOP_CONF_DIR=/export/servers/hadoop-2.7.4/etc/hadoop1整合Spark和HDFS。Spark加载HDFS上的文件，需要修改spark-env.sh配置文件，添加HADOOP_CONF_DIR配置参数，指定Hadoop配置文件的目录，添加配置参数如下。SparkShell读取HDFS文件

#启动Hadoop集群start-all.sh#启动zookeepercd~zk_start.sh#启动spark集群/export/servers/spark/sbin/start-all.sh配置完毕后，启动Hadoop集群服务，并重新启动Spark集群服务，使配置文件生效。2启动Hadoop、Spark服务。SparkShell读取HDFS文件

#启动Spark-Shell交互式界面cd/export/servers/spark/bin/spark-shell--masterlocal[2]#Spark-Shell本身就是一个Driver，它会初始化一个SparkContext对象为“sc”，用户可以直接调用

scala>sc.textFile("/spark/test/words.txt").flatMap(_.split("")).map((_,1)).reduceByKey(_+_).collectres0:Array[(String,Int)]=Array((yuanlongping,1),(hello,4),(spark,1),(huangxuhua,1))3启动Spark–Shell编写程序。SparkShell读取HDFS文件

#可以使用命令：quit或者ctrl+D退出Spark-shellScala>:quit4退出Spark-Shell客户端。66小结掌握SparkShell的启动掌握SparkShell和HDFS交互通信数据分析与实战Spark基础第二章第6节68掌握创建Maven项目掌握IDEA编写Spark程序流程掌握Spark集群运行Spark程序学习目标TARGET

Spark-Shell通常在测试和验证我们的程序时使用的较多，然而在生产环境中，通常会在IDEA开发工具中编写程序，然后打成Jar包，最后提交到集群中执行。本节我们将利用IDEA工具开发一个WordCount单词计数程序。

WordCount案例步骤一：创建Maven项目步骤二：添加Spark相关依赖，打包插件步骤三：编写代码，查看结果WordCount案例创建一个名为“spark_chapter02”的Maven项目，然后在main和test目录下分别创建一个名称为scala的文件夹。步骤一：创建Maven项目将文件夹标记为资源文件夹类型WordCount案例步骤二：添加Spark相关依赖，打包插件设置依赖的版本号添加了Scala、Hadoop和Spark相关的依赖根据素材pom.xml复制内容WordCount案例步骤三：编写代码，查看结果在main目录的scala文件夹中，创建WordCount.scala文件实现词频统计。

objectWordCount{

defmain(args:Array[String]):Unit={

//1.创建SparkConf对象,设置appName和Master地址

valsparkconf=new

SparkConf().setAppName("WordCount").setMaster("local[2]")

//2.创建SparkContext对象，它是所有任务计算的源头，

//它会创建DAGScheduler和TaskScheduler

valsparkContext=newSparkContext(sparkconf)

//3.读取数据文件,RDD可以简单的理解为是一个集合-集合中存放的元素是String类型

valdata:RDD[String]=sparkContext.textFile("D:\\word\\words.txt")WordCount案例步骤三：编写代码，查看结果

//4.切分每一行，获取所有的单词

valwords:RDD[String]=data.flatMap(_.split(""))

//5.每个单词记为1，转换为（单词，1）

valwordAndOne:RDD[(String,Int)]=words.map(x=>(x,1))

//6.相同单词汇总，前一个下划线表示累加数据，后一个下划线表示新数据

valresult:RDD[(String,Int)]=wordAndOne.reduceByKey(_+_)

//7.收集打印结果数据

valfinalResult:Array[(String,Int)]=result.collect()println(finalResult.toBuffer)

//8.关闭sparkContext对象

sparkContext.stop()集群模式执行Spark程序步骤一：添加打包插件步骤二：修改代码，打包程序

步骤三：执行提交命令“spark-submit”集群模式执行Spark程序步骤一：添加打包插件在实际工作应用中，代码编写完成后，需要将程序打包，上传至服务器运行，因此还需要向pom.xml文件中添加所需插件，具体配置参数见素材中pom.xmlbuild中的内容小提示：如果在创建Maven工程中选择Scala原型模板，上述插件会自动创建。这些插件的主要功能是方便开发人员进行打包。集群模式执行Spark程序在打包项目之前，需要对词频统计的代码进行修改，创建WordCount_Online.scala文件