《大数据技术实战案例教程》课件教学课件-第8章Spark集群部署和基本编程

【第7章回顾】1.了解Kafka功能、来源和特点。2.理解Kafka体系架构及Broker、Producer和Customer三种角色功能。（重点）3.理解Kafka部署要点包括运行环境、运行模式、配置文件perties。（重点）4.掌握KafkaShell命令的使用。（重点）5.熟练掌握在Linux环境下部署Kafka集群，使用KafkaShell命令。（重点）【课程内容】开学第一课（理论1学时）第1章部署全分布模式Hadoop集群（理论3学时+实验2学时）第2章HDFS实战（理论4学时+实验2学时）第3章MapReduce编程（理论4学时+实验2学时）第4章部署ZooKeeper集群和ZooKeeper实战（理论4学时+实验2学时）第5章部署本地模式Hive和Hive实战（理论4学时+实验2学时）第6章Flume实战（理论4学时+实验2学时）第7章Kafka实战（理论4学时+实验2学时）第8章Spark集群部署和基本编程（理论4学时+实验2学时）第9章Flink集群部署和基本编程（选修）第8章Spark集群部署和基本编程8.1初识Spark8.2Spark生态系统（重点）8.3Spark运行架构（重点）8.4RDD的设计与操作（重点，难点）8.5Spark部署要点（重点）8.6Spark接口（重点，难点）8.7综合实战：Spark集群部署和基本编程（实验）8.1初识Spark8.1.1Spark简介8.1.2Spark对比HadoopMapReduce8.1.1Spark简介Spark是基于内存计算的大数据并行计算框架。Spark基于内存计算的特性，减少了迭代计算I/O开销，提高了在大数据环境下数据处理的实时性，同时保证了高容错性、高可伸缩性，允许用户将Spark部署在大量的廉价硬件之上形成集群，提高了并行计算能力。Spark于2009年诞生于美国加州大学伯克利分校AMP实验室，AMP实验室的研究人员发现在机器学习迭代算法场景下，HadoopMapReduce表现得效率低下，为了迭代算法和交互式查询两种典型的场景，MateiZaharia和合作伙伴开发了Spark系统的最初版本。2009年Spark论文发布，2010年Spark正式开源，2013年伯克利将其捐赠给Apache基金会，2014年成为Apache基金会的顶级项目，目前已广泛应用于工业界。Spark于2014年打破了Hadoop保持的基准排序纪录，它使用206个节点在23分钟里完成了100TB数据的排序，而Hadoop使用2000个节点在72分钟里完成同样数据的排序。也就是说，Spark仅使用了1/10Hadoop的计算资源，获得了Hadoop的3倍速度。新纪录的诞生，使得Spark获得多方追捧，也表明了Spark可以作为一个更加快速、高效的大数据计算平台。8.1.1Spark简介优点：（1）运行速度快。Spark基于内存的运行速度是HadoopMapReduce的100倍，基于磁盘的运行速度也是HadoopMapReduce的10倍；Spark程序运行是基于线程模型，以线程方式运行作业，要远比基于进程模型的HadoopMapReduce运行作业资源开销小；Spark使用先进的有向无环图（DirectedAcyclicGraph，DAG）执行引擎，可以优化作业执行，提高作业执行效率。（2）易用性。Spark支持使用Scala、Java、Python、R和SQL语言进行编程，简洁的API设计使用户可以快速构建并行程序，并且Spark支持通过SparkShell进行交互式的Python、Scala等编程，可以方便地在这些Shell中使用Spark集群来验证解决问题的方法。（3）通用性。Spark提供了完整且强大的技术栈，包括SQL查询、流式计算、机器学习和图计算，这些组件可以无缝整合到同一个应用中，足以应对复杂的计算。（4）容错性。Spark引进了弹性分布式数据集（ResilientDistributedDataset，RDD），它是分布在一组节点中的只读对象集合。若丢失了一部分对象集合，Spark可以根据父RDD对它们进行计算，另外可以将数据持久化，从而实现容错。8.1.2Spark对比HadoopMapReduceHadoopMapReduce计算框架主要存在以下缺点：（1）表达能力有限。计算都必须转化为Map和Reduce两个操作，但这并不适合所有的情况，难以描述复杂的数据处理过程。（2）磁盘I/O开销大。每次执行时都需要从磁盘读取数据，并且在计算完成后需要将中间结果写入磁盘，I/O开销较大。（3）延迟高。一次计算可能需要分解成一系列按顺序执行的MapReduce任务，任务之间的衔接由于涉及I/O开销，会产生较高延迟。而且，在前一个任务执行完成之前，其它任务无法开始，因此，无法胜任多阶段的计算任务。8.1.2Spark对比HadoopMapReduce相比HadoopMapReduce，Spark主要具有以下优点：（1）Spark计算模型也属于MapReduce，但不局限于Map和Reduce操作，其提供了多种数据集操作类型，编程模型比MapReduce更为灵活。（2）Spark提供了内存计算，可以将计算数据、中间结果直接存放在内存中，大大减少了I/O开销，带来了更高的迭代计算效率，Spark更适合迭代计算较多的机器学习运算。（3）Spark采用了基于DAG的任务调度执行机制，要优于MapReduce的迭代执行机制。HadoopMapReduce与Spark的执行流程对比执行计算读取数据执行计算读取数据产生中间结果输出结果文件存储系统本地磁盘文件存储系统内存读取数据执行计算产生中间结果输出结果文件存储系统（b）Spark执行流程文件存储系统执行计算内存读取数据读取数据（a）HadoopMapReduce执行流程Hadoop与Spark执行逻辑回归的时间对比8.1.2Spark对比HadoopMapReduce尽管Spark相对于Hadoop而言具有较大优势，但Spark并不能完全替代Hadoop，Spark主要用于替代Hadoop中的MapReduce计算模型，Hadoop生态系统中的HDFS和YARN仍是许多大数据体系的核心架构。Spark已经很好地融入了Hadoop生态圈，并成为其中的重要一员，它可以借助YARN实现资源调度管理，借助HDFS实现分布式存储。8.2Spark生态系统SparkSQLSparkStreamingMLlib(machinelearning)GraphX(graph)ApacheSpark

HDFSTachyonYARNMesos应用接口层计算引擎层集群资源管理层存储层StandaloneKubernetes8.2Spark生态系统SparkCore是Spark的核心基础，包含了弹性分布式数据集（RDD）等核心组件。但需要注意的是，SparkCore是离线计算的，这点类似于MapReduce的处理过程。SparkStreaming则是将连续的数据转换为不连续的离散流（DStream），从而实现了快速的数据处理功能。SparkSQL则用于简化Spark库，就好比可以使用Hive简化MapReduce一样，我们也可以使用SparkSQL快速实现Spark开发，具体地讲，SparkSQL可以将DataFrame转为Spark处理时的RDD，然后运行RDD程序。MLlib是Spark上的一个机器学习库，其设计目标是开发一套高可用、高扩展的并行机器学习库，以方便用户直接调用。为方便用户使用，MLlib还提供了一套实用工具集。GraphX是Spark上一个图处理和图并行化计算的组件。用户只需要将样本数据填充到I/O类，然后直接调用图算法，即可完成图的并行化计算。图计算主要应用于社交网络分析等场景。8.3Spark运行架构Driver

SparkContextClusterManagerWorkerNode

Excutor

TaskTaskCacheWorkerNode

Excutor

TaskTaskCache8.3Spark运行架构SparkCore的运行架构采用了标准的Master/Slave架构，Driver为Master，负责管理整个集群中作业任务调度，Executor为Slave，负责执行任务。Spark运行架构包括集群资源管理器（ClusterManager）、运行作业任务的工作节点（WorkerNode）、每个应用的任务控制节点（Driver）和每个工作节点上负责具体任务的执行进程（Executor）。其中，集群资源管理器可以是Spark自带的资源管理器，也可以是YARN或Mesos等第三方资源管理框架。8.4RDD的设计与操作8.4.1RDD简介8.4.2RDD数据存储模型8.4.3RDD创建8.4.4RDD操作8.4.1RDD简介Spark的核心数据模型是RDD，一个RDD就是一个分布式对象集合，本质上是一个只读的分区记录集合，每个RDD可分成多个分区，每个分区就是一个数据集片段，并且一个RDD的不同分区可以被保存到集群中不同的节点上，从而可以在集群中的不同节点上进行并行计算。8.4.1RDD简介可以从以下三个方面来理解RDD。（1）数据集：抽象地说，RDD是一种元素集合。单从逻辑上的表现来看，RDD是一个数据集合，可以简单地将RDD理解为Java中的List集合或数据库中的一张表。（2）分布式：RDD是可以分区的，每个分区可以分布在集群中不同的节点上，从而可以对RDD中的数据进行并行操作。（3）弹性：RDD默认情况下存放在内存中，但是当内存资源不足时Spark会自动将RDD数据写入磁盘进行保存。对于用户来说，不必知道RDD的数据是存储在内存还是磁盘，因为这些都有Spark底层来完成，用户只需针对RDD来进行计算和处理。RDD自动进行内存和磁盘之间权衡和切换的机制是基于RDD弹性的特点。8.4.2RDD数据存储模型Node1

p1p2Node2

p3p6Node3

p4p7Node4

p5p8RDD1RDD28.4.2RDD数据存储模型通常RDD很大，会被划分成很多个分区，分别保存在集群中的不同节点上，分区是逻辑概念。在物理上，RDD对象实质上是一个元数据结构，存储这Block、Node等的映射关系以及其它元数据信息。一个RDD就是一组分区，在物理数据存储上，RDD的每个分区对应的就是一个Block，Block可以存储在内存中，当内存不足时存储到磁盘中。8.4.3RDD创建方法1：从文件系统中加载数据创建RDDSpark采用textFile()方法来从文件系统中加载数据创建RDD。该方法把文件的URI作为参数，这个URI可以是：本地文件系统的地址，或者是分布式文件系统HDFS的地址，或者是AmazonS3的地址等。例如，在pyspark交互式环境中，从本地文件系统中加载数据创建RDD的代码如下所示。>>>lines=sc.textFile("file:///usr/local/spark-2.4.7-bin-without-hadoop-scala-2.12/HelloData/file1.txt")上述代码中，lines就是RDD对象，sc是pyspark启动时自动创建的SparkContext对象，其中对象sc在交互式编程环境中可以直接使用，如果是编写Spark独立应用程序，则可以通过以下语句生成SparkContext对象。frompysparkimportSparkConf,SparkContextconf=SparkConf()sc=SparkContext(conf=conf)例如，在pyspark交互式环境中，从分布式文件系统HDFS中加载数据创建RDD的代码如下所示。>>>lines=sc.textFile("hdfs://30:9000/InputData/file1.txt")8.4.3RDD创建方法2：通过并行集合（列表）创建RDD可以调用SparkContext的parallelize()方法，在Driver中一个已经存在的集合（列表）上创建。例如，在pyspark交互式环境中，通过并行集合（列表）创建RDD的代码如下所示。>>>array=[1,2,3,4,5]>>>rdd=sc.parallelize(array)8.4.4RDD操作转换操作（Transformation）行动操作（Action）RDD的转换操作是返回一个新的RDD的操作，而行动操作则是向Driver返回结果或把结果写入外部系统的操作，会触发实际的计算。1.转换操作操作含义filter(func)筛选出满足函数func的元素，并返回一个新的数据集map(func)将每个元素传递到函数func中，并将结果返回为一个新的数据集flatMap(func)与map()相似，但每个输入元素都可以映射到0或多个输出结果groupByKey()应用于(K,V)键值对的数据集时，返回一个新的(K,Iterable)形式的数据集reduceByKey(func)应用于(K,V)键值对的数据集时，返回一个新的(K,V)形式的数据集，其中每个值是将每个key传递到函数func中进行聚合后的结果2.行动操作操作含义count()返回数据集中的元素个数collect()以数组的形式返回数据集中的所有元素first()返回数据集中的第一个元素take(n)以数组的形式返回数据集中的前n个元素reduce(func)通过函数func（输入两个参数并返回一个值）聚合数据集中的元素foreach(func)将数据集中的每个元素传递到函数func中运行【实例】>>>words=['hello','world','hello','spark']>>>wordsRDD=sc.parallelize(words)>>>pairRDD=wordsRDD.map(lambdax:(x,1))>>>pairRDD.foreach(print)('hello',1)('spark',1)('hello',1)('world',1)['hello','world','hello','spark']words

'hello''world''hello''spark'

('hello',1)('world',1)('hello',1)('spark',1)sc.parallelize(words)wordsRDDwordsRDD.map(lambdax:(x,1))pairRDD8.5Spark部署要点8.5.1Spark运行环境8.5.2Spark运行模式8.5.3Spark配置文件8.5.1Spark运行环境1）操作系统Spark支持Windows和类UNIX（例如Linux、MacOS）操作系统。编者采用的操作系统为Linux发行版CentOS7。2）Java环境Spark采用Scala语言编写，而Scala语言是基于Java的一个脚本化语言，运行在JVM上，因此Spark运行需要Java环境的支持。Spark2.4.7运行在Java8+、Python2.7+/3.4+和R3.5+环境上。对于ScalaAPI，Spark2.4.7使用Scala2.12。编者采用的Java为OracleJDK1.8。需要说明的是，Spark2.2.0及以后版本不再支持Java7、Python2.6和Hadoop2.6.5之前版本。Spark2.3.0不再支持Scala2.10，Spark3.0不再支持Scala2.11。3）SSHSpark集群若想运行，其运行平台Linux上必须安装SSH，且sshd服务必须运行，只有这样，才能使Spark集群中的主节点与集群中所有节点建立通信。本章选用的CentOS7自带有SSH。另外，若Spark采用HDFS存储数据，或采用YARN管理集群资源，则还需要安装Hadoop。8.5.2Spark运行模式Spark运行模式本地模式locallocal[*]local[K]集群模式Standalone模式：使用自带的独立集群资源管理器SparkonYARN模式：使用HadoopYARN作为集群资源管理器yarn-client：以客户端模式连接YARN集群yarn-cluster：以集群模式连接YARN集群SparkonMesos模式：使用ApacheMesos作为集群资源管理器Sparkon*模式：使用其它集群资源管理器Spark集群部署方式部署工具集群资源管理器手工部署工具部署AmbariClouderaManagerStandalone手工部署SparkStandaloneAmbari部署SparkStandaloneClouderaManager部署SparkStandaloneThirdPlatform模式HadoopYARN手工部署SparkonYARNAmbari部署SparkonYARNClouderaManager部署SparkonYARNApacheMesos手工部署SparkonMesosAmbari部署SparkonMesosClouderaManager部署SparkonMesosKubernetes手工部署SparkonKubernetesAmbari部署SparkonKubernetesClouderaManager部署SparkonKubernetes8.5.3Spark配置文件Spark配置文件数量不多，都存放在${SPARK_HOME}/conf下。spark-env.sh配置参数（部分）参数名说明JAVA_HOME指定Java安装路径SPARK_HOME指定Spark安装路径SPARK_CONF_DIR指定Spark集群配置文件位置，默认为${SPARK_HOME}/confSPARK_LOG_DIR指定Spark日志文件的保存位置，默认为${SPARK_HOME}/logsSPARK_PID_DIR指定Spark守护进程号的保存位置，默认为/tmpSPARK_MASTER_HOST指定Spark集群主节点的主机名或IP地址SPARK_WORKER_CORES指定作业可用的CPU内核数量SPARK_WORKER_MEMORY指定作业可使用的内存容量HADOOP_CONF_DIR指定Hadoop集群配置文件位置YARN_CONF_DIR当使用YARN作为集群资源管理器时，指定YARN集群配置文件位置8.6Spark接口8.6.1SparkWebUI8.6.2SparkShell8.6.3SparkAPI8.6.4其它接口8.6.1SparkWebUISparkWebUI主要面向管理员，从该页面上，管理员可以看到正在执行的和已完成的所有Spark应用程序执行过程中的统计信息，该页面只支持读，不支持写操作。SparkWebUI地址默认为http://MasterIP:8080。8.6.2SparkShellSparkShell主要针对Spark程序员和Spark数据分析师的，通过Shell接口，程序员能够向Spark集群提交Spark应用程序、查看正在运行的Spark应用程序；数据分析师可以通过Shell接口以交互式方式对数据进行分析。8.6.2SparkShell1.集群管理集群管理的命令位于“sbin”目录下，可以通过此类命令启动或关闭集群中的某服务或者整个集群。SparkShell集群管理命令功能说明（部分）集群管理Shell命令功能start-all.sh启动Spark集群，注意该命令在Spark主节点机器上执行，且命令执行的前提条件是主节点到本身和其他从节点ssh时皆无需密钥stop-all.sh关闭Spark集群start-master.sh在Spark主节点机器上执行，启动Master进程start-slaves.sh在Spark主节点机器上执行，启动所有Worker进程start-slave.sh在Spark所有从节点机器上执行，启动Worker进程stop-master.sh在Spark主节点机器上执行，关闭Master进程stop-slaves.sh在Spark主节点机器上执行，关闭所有Worker进程stop-slave.sh在Spark所有从节点机器上执行，关闭Worker进程8.6.2SparkShell2.任务管理任务管理的命令位于“bin”目录下，可以通过此类命令向Spark集群提交、管理Spark应用程序，“cmd”后缀的文件为Windows平台脚本。SparkShell任务管理命令功能说明（部分）任务管理Shell命令功能spark-submit向Spark集群提交Spark应用程序pyspark以交互式方式编写并执行Spark应用程序，且书写语法为PythonsparkR以交互式方式编写并执行Spark应用程序，且书写语法为Rspark-shell以交互式方式编写并执行Spark应用程序，且书写语法为Scalaspark-sql以交互式方式编写并执行SparkSQL，且书写语法为类SQLrun-example运行Spark示例程序，实际上，该脚本内部调用了spark-submit，读者不必掌握该命令spark-submit用法（过长，显示部分）$./bin/spark-submit--helpUsage:spark-submit[options]<appjar|pythonfile|Rfile>[apparguments]Usage:spark-submit--kill[submissionID]--master[spark://...]Usage:spark-submit--status[submissionID]--master[spark://...]Usage:spark-submitrun-example[options]example-class[exampleargs]Options:

--masterMASTER_URL spark://host:port,mesos://host:port,yarn,k8s://https://host:port,orlocal(Default:local[*]).

--deploy-modeDEPLOY_MODE Whethertolaunchthedriverprogramlocally("client")orononeoftheworkermachinesinsidethe cluster("cluster")(Default:client).

--classCLASS_NAME Yourapplication'smainclass(forJava/Scalaapps).--nameNAME Anameofyourapplication.--jarsJARS Comma-separatedlistofjarstoincludeontheandexecutorclasspaths.--packages Comma-separatedlistofmavencoordinatesofjarstoincludeonthedriverandexecutorclasspaths. Willsearchthelocalmavenrepo,thenmavencentralandanyadditionalremoterepositoriesgivenby--repositories. TheformatforthecoordinatesshouldbegroupId:artifactId:version.--exclude-packages Comma-separatedlistofgroupId:artifactId,toexcludewhileresolvingthedependenciesprovidedin--packagestoavoid dependencyconflicts.--repositories Comma-separatedlistofadditionalremoterepositoriestosearchforthemavencoordinatesgivenwith--packages.--py-filesPY_FILES Comma-separatedlistof.zip,.egg,or.pyfilestoplaceonthePYTHONPATHforPythonapps.--filesFILES Comma-separatedlistoffilestobeplacedintheworkingdirectoryofeachexecutor.Filepathsofthesefiles inexecutorscanbeaccessedviaSparkFiles.get(fileName).……spark-submit用法Spark程序由谁来调度执行，是由Spark程序提交时MASTER_URL值决定的。上述“spark-submit”命令中，选项“--masterMASTER_URL”决定了集群资源调度方式：如果该选项的值以spark://开头，则使用Spark自带的集群资源管理器来调度；如果其值是yarn，则是使用YARN来调度。关于选项--master的具体取值如下所示：--masterMASTER_URL spark://host:port,mesos://host:port,yarn,k8s://https://host:port,orlocal(Default:local[*]).<MASTER_URL>参数值含义<MASTER_URL>参数值含义local使用1个Worker线程本地化运行Spark（完全不并行）local[*]使用与逻辑CPU个数相同数量的线程来本地化运行Spark（逻辑CPU个数=物理CPU个数*每个物理CPU包含的CPU核数）local[K]使用K个Worker线程本地化运行Spark（理想情况下，K应该根据运行机器的CPU核数设定）spark://HOST:PORTSpark采用Standalone集群模式，连接到指定的Spark集群，默认端口是7077yarn-clientSpark采用YARN集群模式，以客户端模式连接YARN集群，集群的位置可以在HADOOP_CONF_DIR环境变量中找到。用户提交作业后，不能关掉Client，DriverProgram驻留在Client中，适合运行交互类型作业，常用于开发测试yarn-clusterSpark采用YARN集群模式，以集群模式连接YARN集群，集群的位置可以在HADOOP_CONF_DIR环境变量中找到。用户提交作业后，可以关掉Client，作业继续在YARN上执行，不适合运行交互类型作业，常用于企业生产环境mesos://HOST:PORTSpark采用Mesos集群模式，连接到指定的Mesos集群，默认接口是5050pyspark用法（过长，显示部分，与spark-submit类似）$./bin/pyspark--helpUsage:./bin/pyspark[options]Options:--masterMASTER_URL spark://host:port,mesos://host:port,yarn,k8s://https://host:port,orlocal(Default:local[*]).--deploy-modeDEPLOY_MODE Whethertolaunchthedriverprogramlocally("client")orononeoftheworkermachinesinsidethecluster("cluster") (Default:client).--classCLASS_NAME Yourapplication'smainclass(forJava/Scalaapps).--nameNAME Anameofyourapplication.--jarsJARS Comma-separatedlistofjarstoincludeonthedriverandexecutorclasspaths.--packages Comma-separatedlistofmavencoordinatesofjarstoincludeonthedriverandexecutorclasspaths.Willsearchthelocal mavenrepo,thenmavencentralandanyadditionalremoterepositoriesgivenby--repositories.Theformatforthe coordinatesshouldbegroupId:artifactId:version.--exclude-packages Comma-separatedlistofgroupId:artifactId,toexcludewhileresolvingthedependenciesprovidedin--packagestoavoid dependencyconflicts.--repositories Comma-separatedlistofadditionalremoterepositoriestosearchforthemavencoordinatesgivenwith--packages.--py-filesPY_FILES Comma-separatedlistof.zip,.egg,or.pyfilestoplaceonthePYTHONPATHforPythonapps.--filesFILES Comma-separatedlistoffilestobeplacedintheworkingdirectoryofeachexecutor.Filepathsofthesefiles inexecutorscanbeaccessedviaSparkFiles.get(fileName).……8.6.3SparkAPISparkAPI面向Java、Scala、Python、R、SQL工程师和分析师，程序员可以通过这些接口编写Spark应用程序用户层代码ApplicationBusinessLogic。APIDocs网址SparkScalaAPI(Scaladoc)/docs/latest/api/scala/org/apache/spark/index.htmlSparkJavaAPI(Javadoc)/docs/latest/api/java/index.htmlSparkPythonAPI(Sphinx)/docs/latest/api/python/index.htmlSparkRAPI(Roxygen2)/docs/latest/api/R/index.htmlSparkSQL,Built-inFunctions(MkDocs)/docs/latest/api/sql/index.html8.6.4其它接口SparkSQLSparkStreamingSparkMllibSparkGraphX参考官网/docs/latest

8.7综合实战：Spark集群部署和基本编程规划Spark集群部署Spark集群启动Spark集群验证Spark集群编写并运行Spark应用程序关闭Spark集群8.7.1规划Spark集群主机名IP地址运行服务软硬件配置master30Master（Spark主进程）NameNode（HDFS主进程）SecondaryNameNode（与NameNode协同工作）内存：4GCPU：1个2核硬盘：40G操作系统：CentOS7.6.1810Java：OracleJDK8u191Hadoop：Hadoop2.9.2Python：Python3.6.7Spark：Spark2.4.7slave131Worker（Spark从进程）DataNode（HDFS从进程）内存：1GCPU：1个1核硬盘：20G操作系统：CentOS7.6.1810Java：OracleJDK8u191Hadoop：Hadoop2.9.2Python：Python3.6.7Spark：Spark2.4.7slave232Worker（Spark从进程）DataNode（HDFS从进程）内存：1GCPU：1个1核硬盘：20G操作系统：CentOS7.6.1810Java：OracleJDK8u191Hadoop：Hadoop2.9.2Python：Python3.6.7Spark：Spark2.4.78.7.1规划Spark集群软件名称软件版本发布日期下载地址VMwareWorkstationProVMwareWorkstation14.5.7ProforWindows2017年6月22日/products/workstation-pro.htmlCentOSCentOS7.6.18102018年11月26日/download/JavaOracleJDK8u1912018年10月16日/technetwork/java/javase/downloads/index.htmlHadoopHadoop2.9.22018年11月19日/releases.htmlPythonPython3.6.72018年10月20日/downloads/source/SparkSpark2.4.72020年9月12日/downloads.html8.7.2部署Spark集群1.初始软硬件环境准备（1）准备三台机器，安装操作系统，编者使用CentOS7.6.1810。（2）对集群内每一台机器，配置静态IP、修改机器名、添加集群级别域名映射、关闭防火墙。（3）对集群内每一台机器，安装和配置Java，要求Java8或更高版本，编者使用OracleJDK8u191。（4）安装和配置Linux集群中各节点间的SSH免密登录。（5）在Linux集群上部署全分布模式Hadoop集群（可选，若Spark底层存储采用HDFS，资源管理器采用YARN，则需要部署Hadoop集群）。（6）对集群内每一台机器，安装和配置Python3。（CentOS7.6.1810自带Python2.7.5，本书使用Python3编写Spark应用程序）安装和配置Python3（1）使用命令“whichpython”查看系统有无安装Python及其位置，使用命令“python-V”查看Python版本。（2）切换到root用户下，安装依赖包。要求计算机联网，依次执行以下命令。[root@masterxuluhui]#yum-ygroupinstall'Developmenttools'[root@masterxuluhui]#yum-yinstallopenssl-develbzip2-develexpat-develgdbm-develreadline-develsqlite-devel[root@masterxuluhui]#yuminstalllibffi-devel-y上述依赖包安装成功后会有“Complete！”提示信息。安装和配置Python3（3）下载Python3。[root@masterxuluhui]#cd/home/xuluhui/Downloads[root@masterDownloads]#wget/ftp/python/3.6.7/Python-3.6.7.tgz（4）解压Python3。使用以下命令将Python解压到例如目录/home/xuluhui/Downloads下。[root@masterDownloads]#tar-zxvfPython-3.6.7.tgz（5）创建编译安装目录。使用以下命令创建目录/usr/local/python3，Python3.6.7将安装到此目录下，当然，读者也可以选择其它目录。[root@masterDownloads]#mkdir/usr/local/python3安装和配置Python3（6）源码编译和安装。[root@masterDownloads]#cd/home/xuluhui/Downloads/Python-3.6.7[root@masterPython-3.6.7]#./configure--prefix=/usr/local/python3--enable-optimizations[root@masterPython-3.6.7]#make&&makeinstall其中第二条命令的第一个参数--prefix指定安装路径，若不指定，安装过程中软件所需要的文件可能会复制到其他不同目录，从而导致删除软件和复制软件都不方便。第二个参数--enable-optimizations可以提高10%-~20%代码运行速度。（7）使用“ln”命令创建软链接，具体命令如下所示。[root@masterPython-3.6.7]#ln-s/usr/local/python3/bin/python3/usr/bin/python3（8）验证Python3，使用命令“python3-V”测试是否安装成功。安装和配置Python3（8）验证Python3，使用命令“python3-V”测试是否安装成功。8.7.2部署Spark集群2.获取SparkSpark官方下载地址为/downloads.html，建议读者到官网下载稳定版，编者采用2020年9月12日发布的稳定版Spark2.4.7，安装包名称为“spark-2.4.7-bin-without-hadoop-scala-2.12.tgz”，属于“Hadoopfree”版，可应用到任意ApacheHadoop版本，例如存放在master机器的/home/xuluhui/Downloads中。3.主节点上安装Spark并设置属主在master机器上，使用root用户解压spark-2.4.7-bin-without-hadoop-scala-2.12.tgz到安装目录如/usr/local下，依次使用的命令如下所示。[xuluhui@master~]$suroot[root@masterxuluhui]#cd/usr/local[root@masterlocal]#tar-zxvf/home/xuluhui/Downloads/spark-2.4.7-bin-without-hadoop-scala-2.12.tgz为了在普通用户下使用Spark集群，将Spark安装目录的属主设置为Linux普通用户例如xuluhui，使用以下命令完成。[root@masterlocal]#chown-Rxuluhui/usr/local/spark-2.4.7-bin-without-hadoop-scala-2.128.7.2部署Spark集群4.主节点上配置Spark1）配置slaves（1）使用命令“cpslaves.templateslaves”复制模板配置文件slaves.template并命名为“slaves”。（2）使用命令“vimslaves”编辑配置文件slaves，将其中的“localhost”替换为如下内容。slave1slave28.7.2部署Spark集群4.主节点上配置Spark2）配置spark-env.sh环境配置文件spark-env.sh用于指定Spark运行时的各种参数，主要包括Java安装路径JAVA_HOME、Spark安装路径SPARK_HOME等，另外，Spark集群若要存取HDFS文件，必须设置Spark使用HDFS，需要配置Hadoop集群配置文件位置HADOOP_CONF_DIR。（1）使用命令“cpspark-env.sh.templatespark-env.sh”复制模板配置文件spark-env.sh.template并命名为“spark-env.sh”。（2）使用命令“vimspark-env.sh”编辑配置文件spark-env.sh，将以下内容追加到文件最后。exportJAVA_HOME=/usr/java/jdk1.8.0_191exportSPARK_HOME=/usr/local/spark-2.4.7-bin-without-hadoop-scala-2.12exportSPARK_MASTER_HOST=masterexportSPARK_PID_DIR=${SPARK_HOME}/pidsexportPYSPARK_PYTHON=python3exportSPARK_DIST_CLASSPATH=$(/usr/local/hadoop-2.9.2/bin/hadoopclasspath)exportHADOOP_CONF_DIR=/usr/local/hadoop-2.9.2/etc/hadoop8.7.2部署Spark集群4.主节点上配置Spark2）配置spark-env.sh参数SPARK_DIST_CLASSPATH是否要设置取决于下载的Spark是否为带Hadoop的版本，由于本章选择下载的是不带Hadoop的Spark版本，因此必须设置该参数，否则会在启动时出现“AJNIerrorhasoccurred,pleasecheckyourinstallationandtryagain”问题。另外，参数HADOOP_CONF_DIR非必须，只有Spark集群存取HDFS文件时才需要配置。参数SPARK_PID_DIR用于指定Spark守护进程号的保存位置，默认为“/tmp”，由于该文件夹用以存放临时文件，系统定时会自动清理，因此本书将“SPARK_PID_DIR”设置为Spark安装目录下的目录pids，其中目录pids会随着Spark守护进程的启动而由系统自动创建，无需用户手工创建。若使用/tmp作为Spark守护进程号的保存位置，则需要将此目录的写权限赋予给普通用户xuluhui，否则在启动Spark集群时会出现“PermissionDenied”问题。8.7.2部署Spark集群5.同步Spark文件至所有从节点并设置属主切换到root下，使用“scp”命令将master机器中配置好的目录“spark-2.4.7-bin-without-hadoop-scala-2.12”及下属子目录和文件全部拷贝至所有从节点slave1和slave2上，依次使用的命令如下所示，以slave1为例，使用的命令如下所示。[root@masterconf]#scp-r/usr/local/spark-2.4.7-bin-without-hadoop-scala-2.12root@slave1:/usr/local/spark-2.4.7-bin-without-hadoop-scala-2.12[root@masterconf]#scp-r/usr/local/spark-2.4.7-bin-without-hadoop-scala-2.12root@slave2:/usr/local/spark-2.4.7-bin-without-hadoop-scala-2.12然后，依次将所有从节点slave1、slave2上的Spark安装目录的属主也设置为Linux普通用户例如xuluhui，如在slave1机器上使用以下命令完成。[root@slave1xuluhui]#chown-Rxuluhui/usr/local/spark-2.4.7-bin-without-hadoop-scala-2.128.7.3启动Spark集群在master机器上使用命令“start-all.sh”启动集群。注意，此命令只能在master机器上执行，不可以在其他slave机器和客户端机器上执行，脚本默认本机即为Spark主节点。8.7.4验证Spark集群1.验证进程（方法1）8.7.4验证Spark集群2.验证SparkWebUI（方法2）8.7.4验证Spark集群3.验证提交Spark应用程序（方法3）通过SparkShell命令“spark-submit”或“run-example”向Spark集群提交Spark应用程序。（1）通过run-example提交SparkPi到Spark集群运行并通过终端查看运行过程参数“--masterspark://master:7077”指定了Spark采用Standalone集群模式。[xuluhui@masterspark-2.4.7-bin-without-hadoop-scala-2.12]$./bin/run-example\>--masterspark://master:7077\>SparkPi8.7.4验证Spark集群3.验证提交Spark应用程序（方法3）（1）通过run-example提交SparkPi到Spark集群运行并通过终端查看运行过程也可以通过指定参数“---masteryarn--deploy-modeclient”，使得SparkPi在SparkonYARN客户端模式（YARNClient）下运行。[xuluhui@masterspark-2.4.7-bin-without-hadoop-scala-2.12]$./bin/run-example\>--masteryarn\>--deploy-modeclient\>SparkPi8.7.4验证Spark集群3.验证提交Spark应用程序（方法3）（1）通过run-example提交SparkPi到Spark集群运行并通过终端查看运行过程也可以通过指定参数“---masteryarn--deploy-modecluster”，使得SparkPi在SparkonYARN集群模式（YARNCluster）下运行。[xuluhui@masterspark-2.4.7-bin-without-hadoop-scala-2.12]$./bin/run-example\>--masteryarn\>--deploy-modecluster\>SparkPi使用SparkonYARN集群模式（YARNCluster）运行Spark-App时，在终端窗口看不到程序的输出结果8.7.4验证Spark集群3.验证提交Spark应用程序（方法3）（2）通过SparkWebUI或YARNWebUI查看SparkPi运行过程8.7.5编写并运行Spark应用程序【案例8-1】使用SparkShell命令“pyspark”进行交互式编程，使用Python语言编写代码，实现对HDFS文件的英文词频统计，并按词频降序排序，最后输出结果。要求通过终端和SparkWebUI观察该应用程序的运行过程。【案例8-1】（1）使用“pyspark”命令连接到Spark集群。在Linux终端下输入以下命令启动pyspark环境，其中参数“--masterspark://master:7077”指定了Spark采用Standalone集群模式。[xuluhui@masterspark-2.4.7-bin-without-hadoop-scala-2.12]$./bin/pyspark--masterspark://master:7077【案例8-1】（1）使用“pyspark”命令连接到Spark集群。也可以通过指定pyspark参数“--masteryarn--deploy-modeclient”或“--masteryarn--deploy-modecluster”，使得Spark应用程序在SparkonYARN模式上运行。例如以下命令使得Spark应用程序通过pyspark在SparkonYARN客户端模式（YARNClient）上运行。[xuluhui@masterspark-2.4.7-bin-without-hadoop-scala-2.12]$./bin/pyspark--masteryarn--deploy-modeclient例如以下命令使得Spark应用程序通过pyspark在SparkonYARN集群模式（YARNCluster）上运行。[xuluhui@masterspark-2.4.7-bin-without-hadoop-scala-2.12]$./bin/pyspark--masteryarn--deploy-modecluster【案例8-1】（2）使用Python进行交互式编程，实现对HDFS文件的英文词频统计。假设HDFS目录/InputData下已有2个文件file1.txt和file2.txt，file1.txt内容为：HelloHadoopHelloHDFSHelloXijingUniversityfile2.txt内容为：HelloSparkHelloFlinkHelloXijingUniversity【案例8-1】（2）使用Python进行交互式编程，实现对HDFS文件的英文词频统计。第1条语句，sc.textFile()完成了由HDFS文件创建出RDD对象wordRDD，该WordRDD共有6个字符串元素，分别为'HelloHadoop'、'HelloHDFS'、'HelloXijingUniversity'、'HelloSpark'、'HelloFlink'、'HelloXijingUniversity'。第2条语句，首先通过flatMap(lambdax:x.split(""))将RDD对象wordRDD中每个元素按空格进行分割，得到中间RDD1，其共有14个元素，分别为'Hello'、'Hadoop'、'Hello'、'HDFS'、'Hello'、'Xijing'、'University'、'Hello'、'Spark'、'Hello'、'Flink'、'Hello'、'Xijing'、'University'；然后通过map(lambdax:(x,1))将中间RDD1转换为键值对(单词,1)形式，得到中间RDD2，其有14个元素，分别为('Hello',1)、('Hadoop',1)、('Hello',1)、('HDFS',1)、('Hello',1)、('Xijing',1)、('University,1)'、('Hello',1)、('Spark',1)、('Hello',1)、('Flink',1)、('Hello',1)、('Xijing',1)、('University',1)；接着通过reduceByKey(lambdax,y:x+y)将中间RDD2所有元素按Key即单词本身进行分组再对值列表中所有值进行累加操作，得到中间RDD3，其有7个元素，分别为('Hello',6)、('Hadoop',1)、('HDFS',1)、('Xijing',2)、('University,2)'、('Spark',1)、('Flink',1)；最后通过sortBy(lambdax:x[1],Flase)将中间RDD3所有元素按x[1]即Value也就是频次进行降序排序，得到RDD对象wordCount，其有7个元素，依次为('Hello',6)、('Xijing',2)、('University',2)、('HDFS',1)、('Spark',1)、('Flink',1)、('Hadoop',1)。第3条语句，通过wordCount.collect()将RDD对象wordCount中的所有元素以元组形式返回，显示在终端。【案例8-1】（3）通过SparkWebUI或YARNWebUI查看PySparkShell运行过程。8.7.5编写并运行Spark应用程序【案例8-2】使用Python语言编写SparkStreaming独立应用程序，将Flume作为SparkStreaming输入源，实现对Flume消息的实时词频统计。要求从终端上不断给FlumeSource（netcat类型）发送各种消息，Flume把消息汇集到Sink（avro类型），由Sink把消息推送给SparkStreaming。【案例8-2】（1）创建Agent属性文件perties依次使用的命令如下所示。[xuluhui@master~]$cd/usr/local/flume-1.9.0/conf[xuluhui@masterconf]$vim./perties文件perties的具体内容如下所示。#perties:Asingle-nodeFlumeconfiguration#Namethecomponentsonthisagenta1.sources=r1a1.sinks=k1a1.channels=c1#Describe/configurethesourcea1.sources.r1.type=netcata1.sources.r1.bind=localhosta1.sources.r1.port=33333#Describethesinka1.sinks.k1.type=avroa1.sinks.k1.hostname=localhosta1.sinks.k1.port=44444#Useachannelwhichbufferseventsinmemorya1.channels.c1.type=memorya1.channels.c1.capacity=1000000a1.channels.c1.transactionCapacity=1000000#Bindthesourceandsinktothechannela1.sources.r1.channels=c1a1.sinks.k1.channel=c1【案例8-2】（2）Spark准备工作①下载spark-streaming-flume_2.12-2.4.7.jar，其中2.12对应Scala版本，2.4.7对应Spark版本。下载地址如/artifact/org.apache.spark/spark-streaming-flume_2.12/2.4.7。②把jar文件spark-streaming-flume_2.12-2.4.7.jar放到