typora-copy-images-to: upload
Spark环境搭建
什么是Spark
回顾：Hadoop主要解决，海量数据的存储和海量数据的分析计算。
Spark是一种基于内存的快速、通用、可扩展的大数据分析计算引擎
Spark运行模式
部署Spark集群大体上分为两种模式：单机模式与集群模式
大多数分布式框架都支持单机模式，方便开发者调试框架的运行环境。但是在生产环境中，并不会使用单机模式。因此，后续直接按照集群模式部署Spark集群。
下面详细列举了Spark目前支持的部署模式。
（1）Local模式：在本地部署单个Spark服务
（2）Standalone模式：Spark自带的任务调度模式。（国内常用）
（3）YARN模式：Spark使用Hadoop的YARN组件进行资源与任务调度。（国内最常用）
（4）Mesos模式：Spark使用Mesos平台进行资源与任务的调度。（国内很少用）
Spark安装地址
1）官网地址：http://spark.apache.org/
2）文档查看地址：https://spark.apache.org/docs/3.0.0/
3）下载地址：https://spark.apache.org/downloads.html
https://archive.apache.org/dist/spark/
Local模式
Local模式就是运行在一台计算机上的模式，通常就是用于在本机上练手和测试。
安装使用
1）上传并解压Spark安装包
[atguigu@hadoop102 sorfware]$ tar-zxvf spark-3.0.0-bin-hadoop3.2.tgz -C /opt/module/
[atguigu@hadoop102 module]$ mv spark-3.0.0-bin-hadoop3.2 spark-local
2）官方求PI案例
[atguigu@hadoop102 spark-local]$ bin/spark-submit \--class org.apache.spark.examples.SparkPi \--master local[2]\
./examples/jars/spark-examples_2.12-3.0.0.jar \10
可以查看spark-submit所有参数：
Ø --class：表示要执行程序的主类；
Ø --master local[2]
（1）local: 没有指定线程数，则所有计算都运行在一个线程当中，没有任何并行计算
（2）local[K]:指定使用K个Core来运行计算，比如local[2]就是运行2个Core来执行
（3）local[*]：默认模式。自动帮你按照CPU最多核来设置线程数。比如CPU有8核，
Spark帮你自动设置8个线程计算。
Ø spark-examples_2.12-3.0.0.jar：要运行的程序；
Ø 10：要运行程序的输入参数（计算圆周率π的次数，计算次数越多，准确率越高）；
计算结果
注意：首先要启动hadoop，否则会报错
配置历史服务
由于spark-shell停止掉后，hadoop102:4040页面就看不到历史任务的运行情况，所以开发时都配置历史服务器记录任务运行情况。
1）修改spark-default.conf.template名称
[atguigu@hadoop102 conf]$ mv spark-defaults.conf.template spark-defaults.conf
2）修改spark-default.conf文件，配置日志存储路径（写）
[atguigu@hadoop102 conf]$ vim spark-defaults.conf
spark.eventLog.enabled true
spark.eventLog.dir hdfs://hadoop102:8020/directory
注意：需要启动Hadoop集群，HDFS上的目录需要提前存在。
[atguigu@hadoop102 hadoop-3.1.3]$ sbin/start-dfs.sh
[atguigu@hadoop102 hadoop-3.1.3]$ hadoop fs -mkdir /directory
3）修改spark-env.sh文件，添加如下配置：****
[atguigu@hadoop102 conf]$ vim spark-env.sh
exportSPARK_HISTORY_OPTS="
-Dspark.history.ui.port=18080
-Dspark.history.fs.logDirectory=hdfs://hadoop102:8020/directory
-Dspark.history.retainedApplications=30"
参数1含义：WEBUI访问的端口号为18080
参数2含义：指定历史服务器日志存储路径（读）
参数3含义：指定保存Application历史记录的个数，如果超过这个值，旧的应用程序信息将被删除，这个是内存中的应用数，而不是页面上显示的应用数。
4）分发配置文件
[atguigu@hadoop102 conf]$ xsync spark-defaults.conf spark-env.sh
5）启动历史服务
[atguigu@hadoop102 spark-standalone]$
sbin/start-history-server.sh
6）再次执行任务
查看Spark历史服务地址：hadoop102:18080
集群角色
Master和Worker集群资源管理
Master和Worker是Spark的守护进程、集群资源管理者，即Spark在特定模式(Standalone)下正常运行必须要有的后台常驻进程
Driver和Executor任务的管理者
Standalone模式
Standalone模式是Spark自带的资源调度引擎，构建一个由Master + Worker构成的Spark集群，Spark运行在集群中。
这个要和Hadoop中的Standalone区别开来。这里的Standalone是指只用Spark来搭建一个集群，不需要借助Hadoop的Yarn和Mesos等其他框架。
安装使用
1）集群规划
hadoop102hadoop103hadoop104SparkMaster WorkerWorkerWorker
一台master，两台worker
2）再解压一份Spark安装包，并修改解压后的文件夹名称为spark-standalone
atguigu@hadoop102 sorfware]$ tar-zxvf spark-3.0.0-bin-hadoop3.2.tgz -C /opt/module/
[atguigu@hadoop102 module]$ mv spark-3.0.0-bin-hadoop3.2 spark-standalone
3）进入Spark的配置目录/opt/module/spark-standalone/conf
[atguigu@hadoop102 spark-standalone]$ cd conf
4）修改slave文件，添加work节点：
[atguigu@hadoop102 conf]$ mv slaves.template slaves
[atguigu@hadoop102 conf]$ vim slaves
hadoop102
hadoop103
hadoop104
5）修改spark-env.sh文件，添加master节点
[atguigu@hadoop102 conf]$ mv spark-env.sh.template spark-env.sh
[atguigu@hadoop102 conf]$ vim spark-env.sh
SPARK_MASTER_HOST=hadoop102
SPARK_MASTER_PORT=7077
6）分发spark-standalone包
[atguigu@hadoop102 module]$ xsync spark-standalone/
7）启动spark集群
[atguigu@hadoop102 spark-standalone]$ sbin/start-all.sh
注意：如果遇到 “JAVA_HOME not set” 异常，可以在sbin目录下的spark-config.sh 文件中加入如下配置。
export JAVA_HOME=环境变量
8）网页查看：hadoop102:8080（master web的端口，相当于yarn的8088端口）
9）官方求PI案例
atguigu@hadoop102 spark-standalone]$ bin/spark-submit \--class org.apache.spark.examples.SparkPi \--master spark://hadoop102:7077 \
./examples/jars/spark-examples_2.12-3.0.0.jar \10
参数说明
参数解释可选值举例–classSpark程序中包含主函数的类–masterSpark****程序运行的模式本地模式：local[*]、spark://hadoop102:7077、 Yarn–executor-memory 1G指定每个executor可用内存为1G符合集群内存配置即可，具体情况具体分析。–total-executor-cores 2指定所有executor使用的cpu核数为2个application-jar打包好的应用jar，包含依赖。这个URL在集群中全局可见。 比如hdfs:// 共享存储系统，如果是file:// path，那么所有的节点的path都包含同样的jarapplication-arguments传给main()方法的参数
1）基本语法
bin/spark-submit \--class<main-class>--master<master-url>\... # other options<application-jar>\[application-arguments]
2）配置Executor可用内存为2G，使用CPU核数为2个
[atguigu@hadoop102 spark-standalone]$ bin/spark-submit \--class org.apache.spark.examples.SparkPi \--master spark://hadoop102:7077 \
--executor-memory 2G \
--total-executor-cores 2\
./examples/jars/spark-examples_2.12-3.0.0.jar \10
配置高可用（HA）
（1）Zookeeper正常安装并启动
[atguigu@hadoop102 zookeeper-3.4.10]$ zk.sh start
（2）修改spark-env.sh文件添加如下配置：
atguigu@hadoop102 conf]$ vim spark-env.sh
#注释掉如下内容：#SPARK_MASTER_HOST=hadoop102#SPARK_MASTER_PORT=7077#添加上如下内容。配置由Zookeeper管理Master，在Zookeeper节点中自动创建/spark目录，用于管理：exportSPARK_DAEMON_JAVA_OPTS="
-Dspark.deploy.recoveryMode=ZOOKEEPER
-Dspark.deploy.zookeeper.url=hadoop102,hadoop103,hadoop104
-Dspark.deploy.zookeeper.dir=/spark"#添加如下代码#Zookeeper3.5的AdminServer默认端口是8080，和Spark的WebUI冲突exportSPARK_MASTER_WEBUI_PORT=8989
（3）分发配置文件
[atguigu@hadoop102 conf]$ xsync spark-env.sh
（4）在hadoop102上启动全部节点
[atguigu@hadoop102 spark-standalone]$ sbin/start-all.sh
（5）在hadoop103上单独启动master节点
[atguigu@hadoop103 spark-standalone]$ sbin/start-master.sh
（6）在启动一个hadoop102窗口，将/opt/module/spark-local/input数据上传到hadoop集群的/input目录
[atguigu@hadoop102 spark-standalone]$ hadoop fs -put /opt/module/spark-local/input/ /input
（7）Spark HA集群访问
[atguigu@hadoop102 spark-standalone]$
bin/spark-shell \--master spark://hadoop102:7077,hadoop103:7077 \
--executor-memory 2g \
--total-executor-cores 2
参数：–master spark://hadoop102:7077指定要连接的集群的master。
注：一旦配置了高可用以后，master后面要连接多个master。
Yarn模式
Spark客户端直接连接Yarn，不需要额外构建Spark集群。
安装使用
0）停止Standalone模式下的spark集群
[atguigu@hadoop102 spark-standalone]$ sbin/stop-all.sh
[atguigu@hadoop102 spark-standalone]$ zk.sh stop
[atguigu@hadoop103 spark-standalone]$ sbin/stop-master.sh
1）为了防止和Standalone模式冲突，再单独解压一份spark
[atguigu@hadoop102 software]$ tar-zxvf spark-3.0.0-bin-hadoop3.2.tgz -C /opt/module/
2）进入到/opt/module目录，修改spark-3.0.0-bin-hadoop3.2名称为spark-yarn
[atguigu@hadoop102 module]$ mv spark-3.0.0-bin-hadoop3.2/ spark-yarn
3）修改hadoop配置文件/opt/module/hadoop-3.1.3/etc/hadoop/yarn-site.xml，添加如下内容
因为测试环境虚拟机内存较少，防止执行过程进行被意外杀死，做如下配置
[atguigu@hadoop102 hadoop]$ vim yarn-site.xml
<!--是否启动一个线程检查每个任务正使用的物理内存量，如果任务超出分配值，则直接将其杀掉，默认是true --><property><name>yarn.nodemanager.pmem-check-enabled</name><value>false</value></property><!--是否启动一个线程检查每个任务正使用的虚拟内存量，如果任务超出分配值，则直接将其杀掉，默认是true --><property><name>yarn.nodemanager.vmem-check-enabled</name><value>false</value></property>
4）分发配置文件
[atguigu@hadoop102 conf]$ xsync /opt/module/hadoop-3.1.3/etc/hadoop/yarn-site.xml
5）修改/opt/module/spark-yarn/conf/spark-env.sh，添加YARN_CONF_DIR配置，保证后续运行任务的路径都变成集群路径
[atguigu@hadoop102 conf]$ mv spark-env.sh.template spark-env.sh
[atguigu@hadoop102 conf]$ vim spark-env.sh
YARN_CONF_DIR=/opt/module/hadoop-3.1.3/etc/hadoop
6）启动HDFS以及YARN集群
[atguigu@hadoop102 hadoop-3.1.3]$ sbin/start-dfs.sh
[atguigu@hadoop103 hadoop-3.1.3]$ sbin/start-yarn.sh
7）执行一个程序
[atguigu@hadoop102 spark-yarn]$ bin/spark-submit \--class org.apache.spark.examples.SparkPi \--masteryarn\
./examples/jars/spark-examples_2.12-3.0.0.jar \10
参数：–master yarn，表示Yarn方式运行；–deploy-mode表示客户端方式运行程序
8）如果运行的时候，抛出如下异常
ClassNotFoundException:com.sun.jersey.api.client.config.ClientConfig
-原因分析：Spark2中jersey版本是2.22，但是yarn中还需要依赖1.9，版本不兼容
-解决方式：在yarn-site.xml中，添加
yarn.timeline-service.enabled
false
9）查看hadoop103:8088页面，点击History，查看历史页面
配置历史服务
和Standalone模式配置方法一样
运行流程
Spark有yarn-client和yarn-cluster两种模式，主要区别在于：Driver程序的运行节点。
yarn-client：Driver程序运行在客户端，适用于交互、调试，希望立即看到app的输出。
yarn-cluster：Driver程序运行在由ResourceManager启动的APPMaster，适用于生产环境。
1）客户端模式（默认）
[atguigu@hadoop102 spark-yarn]$ bin/spark-submit \--class org.apache.spark.examples.SparkPi \--masteryarn\
--deploy-mode client \
./examples/jars/spark-examples_2.12-3.0.0.jar \10
2）集群模式
[atguigu@hadoop102 spark-yarn]$ bin/spark-submit \--class org.apache.spark.examples.SparkPi \--masteryarn\
--deploy-mode cluster \
./examples/jars/spark-examples_2.12-3.0.0.jar \10
（1）查看http://hadoop103:8088/cluster页面，点击History按钮，跳转到历史详情页面
（2）http://hadoop102:18080点击Executors->点击driver中的stdout
注意：如果在yarn日志端无法查看到具体的日志，则在yarn-site.xml中添加如下配置并启动Yarn历史服务器
<property><name>yarn.log.server.url</name><value>http://hadoop102:19888/jobhistory/logs</value></property>
注意：hadoop历史服务器也要启动 mr-jobhistory-daemon.sh start historyserver
WordCount案例实操
Spark Shell仅在测试和验证我们的程序时使用的较多，在生产环境中，通常会在IDEA中编制程序，然后打成Jar包，然后提交到集群，最常用的是创建一个Maven项目，利用Maven来管理Jar包的依赖。
编写程序
1）创建一个Maven项目WordCount
2）在项目WordCount上点击右键，Add Framework Support=》勾选scala
3）在main下创建scala文件夹，并右键Mark Directory as Sources Root=>在scala下创建包名为com.atguigu.spark
4）输入文件夹准备：在新建的WordCount项目名称上右键=》新建input文件夹=》在input文件夹上右键=》分别新建1.txt和2.txt。每个文件里面准备一些word单词。
5）导入项目依赖
下方的是scala语言打包插件 只要使用scala语法打包运行到linux上面 必须要有
<dependencies><dependency><groupId>org.apache.spark</groupId><artifactId>spark-core_2.12</artifactId><version>3.0.0</version></dependency></dependencies><build><finalName>WordCount</finalName><plugins><plugin><groupId>net.alchim31.maven</groupId><artifactId>scala-maven-plugin</artifactId><version>3.4.6</version><executions><execution><goals><goal>compile</goal><goal>testCompile</goal></goals></execution></executions></plugin></plugins></build>
注意：如果maven版本为3.2.x，插件下载报错，那么修改插件版本为3.3.2
6）创建伴生对象WordCount，编写代码
packagecom.atguigu.sparkimportorg.apache.spark.rdd.RDD
importorg.apache.spark.{SparkConf, SparkContext}object WordCount {def main(args: Array[String]):Unit={//1.创建SparkConf并设置App名称val conf =new SparkConf().setAppName("WC").setMaster("local[*]")//2.创建SparkContext，该对象是提交Spark App的入口val sc =new SparkContext(conf)//3.读取指定位置文件:hello atguigu atguiguval lineRdd: RDD[String]= sc.textFile("input")//4.读取的一行一行的数据分解成一个一个的单词（扁平化）(hello)(atguigu)(atguigu)val wordRdd: RDD[String]= lineRdd.flatMap(line => line.split(" "))//5. 将数据转换结构：(hello,1)(atguigu,1)(atguigu,1)val wordToOneRdd: RDD[(String,Int)]= wordRdd.map(word =>(word,1))//6.将转换结构后的数据进行聚合处理 atguigu:1、1 =》1+1 (atguigu,2)val wordToSumRdd: RDD[(String,Int)]= wordToOneRdd.reduceByKey((v1, v2)=> v1 + v2)//7.将统计结果采集到控制台打印val wordToCountArray: Array[(String,Int)]= wordToSumRdd.collect()
wordToCountArray.foreach(println)//一行搞定//sc.textFile(args(0)).flatMap(_.split(" ")).map((_, 1)).reduceByKey(_ + _).saveAsTextFile(args(1))//8.关闭连接
sc.stop()}}
Spark程序运行过程中会打印大量的执行日志，为了能够更好的查看程序的执行结果，可以在项目的resources目录中创建log4j.properties文件，并添加日志配置信息：
log4j.rootCategory=ERROR, console
log4j.appender.console=org.apache.log4j.ConsoleAppender
log4j.appender.console.target=System.err
log4j.appender.console.layout=org.apache.log4j.PatternLayout
log4j.appender.console.layout.ConversionPattern=%d{yy/MM/dd HH:mm:ss} %p %c{1}: %m%n
# Set the default spark-shell log level to ERROR. When running the spark-shell, the
# log level for this class is used to overwrite the root logger's log level, so that
# the user can have different defaults for the shell and regular Spark apps.
log4j.logger.org.apache.spark.repl.Main=ERROR
# Settings to quiet third party logs that are too verbose
log4j.logger.org.spark_project.jetty=ERROR
log4j.logger.org.spark_project.jetty.util.component.AbstractLifeCycle=ERROR
log4j.logger.org.apache.spark.repl.SparkIMain$exprTyper=ERROR
log4j.logger.org.apache.spark.repl.SparkILoop$SparkILoopInterpreter=ERROR
log4j.logger.org.apache.parquet=ERROR
log4j.logger.parquet=ERROR
# SPARK-9183: Settings to avoid annoying messages when looking up nonexistent UDFs in SparkSQL with Hive support
log4j.logger.org.apache.hadoop.hive.metastore.RetryingHMSHandler=FATAL
log4j.logger.org.apache.hadoop.hive.ql.exec.FunctionRegistry=ERROR
7）打包插件
<plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-assembly-plugin</artifactId><version>3.0.0</version><configuration><descriptorRefs><descriptorRef>jar-with-dependencies</descriptorRef></descriptorRefs></configuration><executions><execution><id>make-assembly</id><phase>package</phase><goals><goal>single</goal></goals></execution></executions></plugin>
8）打包到集群测试
（1）点击package打包，然后，查看打完后的jar包
（2）将WordCount.jar上传到/opt/module/spark-yarn目录
（3）在HDFS上创建，存储输入文件的路径/input
[atguigu@hadoop102 spark-yarn]$ hadoop fs -mkdir /input
（4）上传输入文件到/input路径
[atguigu@hadoop102 spark-yarn]$ hadoop fs -put /opt/module/spark-local/input/1.txt /input
（5）执行任务
[atguigu@hadoop102 spark-yarn]$ bin/spark-submit \--class com.atguigu.spark.WordCount \--masteryarn\
WordCount.jar \
/input \
/output
注意：input和ouput都是HDFS上的集群路径。
（6）查询运行结果
[atguigu@hadoop102 spark-yarn]$ hadoop fs -cat /output/*
注意：如果运行发生压缩类没找到(如果之前配置了lzo压缩会有这个问题)
（1）原因
Spark on Yarn会默认使用Hadoop集群配置文件设置编码方式，但是Spark在自己的spark-yarn/jars 包里面没有找到支持lzo压缩的jar包，所以报错。
（2）解决方案一：拷贝lzo的包到/opt/module/spark-yarn/jars目录
[atguigu@hadoop102 common]$
cp /opt/module/hadoop-3.1.3/share/hadoop/common/hadoop-lzo-0.4.20.jar /opt/module/spark-yarn/jars
（3）解决方案二：在执行命令的时候指定lzo的包位置
[atguigu@hadoop102 spark-yarn]$
bin/spark-submit \--class com.atguigu.spark.WordCount \--masteryarn\
--driver-class-path /opt/module/hadoop-3.1.3/share/hadoop/common/hadoop-lzo-0.4.20.jar \
WordCount.jar \
/input \
/output
本地调试
本地Spark程序调试需要使用Local提交模式，即将本机当做运行环境，Master和Worker都为本机。运行时直接加断点调试即可。
异常处理
如果本机操作系统是Windows，如果在程序中使用了Hadoop相关的东西，比如写入文件到HDFS，则会遇到如下异常：
出现这个问题的原因，并不是程序的错误，而是用到了Hadoop相关的服务，解决办法
（1）配置HADOOP_HOME环境变量。
（2）在IDEA中配置Run Configuration，添加HADOOP_HOME变量。
t/module/spark-yarn/jars
（3）解决方案二：在执行命令的时候指定lzo的包位置
```shell
[atguigu@hadoop102 spark-yarn]$
bin/spark-submit \
--class com.atguigu.spark.WordCount \
--master yarn \
--driver-class-path /opt/module/hadoop-3.1.3/share/hadoop/common/hadoop-lzo-0.4.20.jar \
WordCount.jar \
/input \
/output
本地调试
本地Spark程序调试需要使用Local提交模式，即将本机当做运行环境，Master和Worker都为本机。运行时直接加断点调试即可。
异常处理
如果本机操作系统是Windows，如果在程序中使用了Hadoop相关的东西，比如写入文件到HDFS，则会遇到如下异常：
[外链图片转存中…(img-JN3E4vAO-1673790782163)]
出现这个问题的原因，并不是程序的错误，而是用到了Hadoop相关的服务，解决办法
（1）配置HADOOP_HOME环境变量。
（2）在IDEA中配置Run Configuration，添加HADOOP_HOME变量。
版权归原作者 爱学习的IT民工 所有， 如有侵权，请联系我们删除。