1. Spark 入门

部署Spark集群大体上分为两种模式：单机模式与集群模式
大多数分布式框架都支持单机模式，方便开发者调试框架的运行环境。但是在生产环境中，并不会使用单机模式。因此，后续直接按照集群模式部署Spark集群。
下面详细列举了Spark目前支持的部署模式。

1. Local模式：在本地部署单个Spark服务
2. Standalone模式：Spark自带的任务调度模式。（国内常用）
3. YARN模式：Spark使用Hadoop的YARN组件进行资源与任务调度。（国内常用）
4. Mesos模式：Spark使用Mesos平台进行资源与任务的调度。

1）官网地址：http://spark.apache.org/

2）文档查看地址：https://spark.apache.org/docs/2.1.1/

3）下载地址：https://spark.apache.org/downloads.html

2. Local模式

上传并解压安装包

cd /opt/software/
tar -zxvf spark-2.1.1-bin-hadoop2.7.tgz -C /opt/module/
cd /opt/module/
mv spark-2.1.1-bin-hadoop2.7/ spark-local
cd spark-local

2.1. 官方求PI案例

利用蒙特·卡罗算法求PI

bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master local[2] \
./examples/jars/spark-examples_2.11-2.1.1.jar \
10

• —class：表示要执行程序的主类；

• —master local[2]

  • local: 没有指定线程数，则所有计算都运行在一个线程当中，没有任何并行计算

  • local[K]:指定使用K个Core来运行计算，比如local[2]就是运行2个Core来执行

  • local[*]: 自动帮你按照CPU最多核来设置线程数。比如CPU有4核，Spark帮你自动设置4个线程计算

• spark-examples_2.11-2.1.1.jar：要运行的程序；

• 10：要运行程序的输入参数（计算圆周率π的次数，计算次数越多，准确率越高）；

2.2. 官方WordCount案例

读取多个输入文件，统计每个单词出现的总次数

准备数据

mkdir input
echo hello world > input/1.txt
echo hello spark > input/2.txt

启动spark-shell

bin/spark-shell

sc是SparkCore程序的入口；spark是SparkSQL程序入口；master = local[*]表示本地模式运行。

sc.textFile("/opt/module/spark-local/input").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).collect

查看web页面

spark-shell窗口关闭掉，则hadoop102:4040页面关闭

本地模式下，默认的调度器为FIFO。

3. 集群角色

3.1. Master 和 Worker

3.2. Driver 和 Executor

Master和Worker是Spark的守护进程，即Spark在特定模式下正常运行所必须的进程。Driver和Executor是临时程序，当有具体任务提交到Spark集群才会开启的程序

4. Standalone模式

Standalone模式是Spark自带的资源调动引擎，构建一个由Master + Slave构成的Spark集群，Spark运行在集群中。

这个要和Hadoop中的Standalone区别开来。这里的Standalone是指只用Spark来搭建一个集群，不需要借助其他的框架。是相对于Yarn和Mesos来说的。

集群规划

解压安装

cd /opt/software/
tar -zxvf spark-2.1.1-bin-hadoop2.7.tgz -C /opt/module/
cd /opt/module/
mv spark-2.1.1-bin-hadoop2.7/ spark-standalone
cd spark-standalone

配置spark

cd conf
mv slaves.template slaves
vim slaves
hadoop102
hadoop103
hadoop104

修改spark-env.sh文件，添加master节点

mv spark-env.sh.template spark-env.sh
vim spark-env.sh
SPARK_MASTER_HOST=hadoop102
SPARK_MASTER_PORT=7077

分发

xsync /opt/module/spark-standalone/

启动集群

cd ..
sbin/start-all.sh

查看进程

jps

================atguigu@hadoop102================
3330 Jps
3238 Worker
3163 Master
================atguigu@hadoop103================
2966 Jps
2908 Worker
================atguigu@hadoop104================
2978 Worker
3036 Jps

如果遇到 “JAVA_HOME not set” 异常，可以在sbin目录下的spark-config.sh 文件中加入如下配置

export JAVA_HOME=/opt/module/jdk1.8.0_212

网页查看 访问hadoop102:8080

官方求PI案例

bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master spark://hadoop102:7077 \
./examples/jars/spark-examples_2.11-2.1.1.jar \
10

查看web页面 hadoop102:8080

一共12个cores 12核 每个核1024内存

我们提交任务时也可以通过属性来控制 核数和内存

配置Executor可用内存为2G，使用CPU核数为2个

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.11-2.1.1.jar \
10

4.1. 配置历史服务

由于spark-shell停止掉后，hadoop102:4040页面就看不到历史任务的运行情况，所以开发时都配置历史服务器记录任务运行情况。

先停止spark

sbin/stop-all.sh

修改spark-default.conf文件

cd /opt/module/spark-standalone/conf/
mv spark-defaults.conf.template spark-defaults.conf
vi spark-defaults.conf

spark.eventLog.enabled          true
spark.eventLog.dir               hdfs://hadoop102:8020/directory

分发

xsync spark-defaults.conf

启动hadoop集群并且保证logdir的目录提前存在

start-dfs.sh
hadoop fs -mkdir /directory

修改spark-env.sh文件

vi spark-env.sh

export SPARK_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历史记录的个数，如果超过这个值，旧的应用程序信息将被删除，这个是内存中的应用数，而不是页面上显示的应用数。

分发配置文件

xsync spark-env.sh

启动spark

sbin/start-all.sh

启动历史服务

sbin/start-history-server.sh

再执行任务

bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master spark://hadoop102:7077 \
--executor-memory 1G \
--total-executor-cores 2 \
./examples/jars/spark-examples_2.11-2.1.1.jar \
10

访问历史服务器

http://hadoop102:18080/

4.2. 配置高可用（HA）

停止spark集群

sbin/stop-all.sh

启动zookeeper

zk.sh start

修改spark-env.sh文件添加如下配置

vim conf/spark-env.sh

#注释掉如下内容：
#SPARK_MASTER_HOST=hadoop102
#SPARK_MASTER_PORT=7077

#添加上如下内容。配置由Zookeeper管理Master，在Zookeeper节点中自动创建/spark目录，用于管理：
export SPARK_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冲突 所以要把spark默认8080端口改为8989
export SPARK_MASTER_WEBUI_PORT=8989

分发配置

xsync conf/spark-env.sh

启动

sbin/start-all.sh

在103上单独启动master节点 一共有两个master节点

sbin/start-master.sh

hadoop103上的master处于待命状态

在102将sprak-local/input 数据上传到hadoop集群的/input目录

hadoop fs -put /opt/module/spark-local/input/ /input

spark HA集群访问 注意master 为两个spark master

bin/spark-shell \
--master spark://hadoop102:7077,hadoop103:7077 \
--executor-memory 2g \
--total-executor-cores 2

执行Wordcount程序

sc.textFile("hdfs://hadoop102:8020/input").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).collect

测试高可用 在102上kill掉Master进程

查看hadoop103:8989 是否从STANDBY 变为Alive

4.3. 运行流程

Spark有standalone-client和standalone-cluster两种模式，主要区别在于：Driver程序的运行节点。

客户端模式

bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master spark://hadoop102:7077,hadoop103:7077 \
--executor-memory 2G \
--total-executor-cores 2 \
--deploy-mode client \
./examples/jars/spark-examples_2.11-2.1.1.jar \
10

—deploy-mode client，表示Driver程序运行在本地客户端 默认为client

集群模式模式

bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master spark://hadoop102:7077,hadoop103:7077 \
--executor-memory 2G \
--total-executor-cores 2 \
--deploy-mode cluster \
./examples/jars/spark-examples_2.11-2.1.1.jar \
10

查看http://hadoop102:8989/页面，点击Completed Drivers里面的Worker的结果

5. Yarn模式

Spark客户端直接连接Yarn，不需要额外构建Spark集群。

停止Standalone模式下的spark集群

sbin/stop-all.sh
sbin/stop-master.sh  #103的master
zk.sh stop

解压安装

cd /opt/software
tar -zxvf spark-2.1.1-bin-hadoop2.7.tgz -C /opt/module/
cd /opt/module/
mv spark-2.1.1-bin-hadoop2.7/ spark-yarn

修改hadoop配置文件/opt/module/hadoop-2.7.2/etc/hadoop/yarn-site.xml 添加如下内容

因为测试环境虚拟机内存较少，防止执行过程进行被意外杀死，做如下配置

vi /opt/module/hadoop-3.1.3/etc/hadoop/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>
<!-- Spark2中jersey版本是2.22，但是yarn中还需要依赖1.9，版本不兼容 -->
<property>
        <name>yarn.timeline-service.enabled</name>
        <value>false</value>
</property>

分发

xsync /opt/module/hadoop-3.1.3/etc/hadoop/yarn-site.xml

修改/opt/module/spark/conf/spark-env.sh，添加YARN_CONF_DIR配置，保证后续运行任务的路径都变成集群路径

cd /opt/module/spark-yarn/conf/
mv spark-env.sh.template spark-env.sh
vi spark-env.sh

YARN_CONF_DIR=/opt/module/hadoop-3.1.3/etc/hadoop

分发spark-yarn

xsync /opt/module/spark-yarn/

启动HDFS以及YARN集群

start-dfs.sh #102
start-yarn.sh #103

执行求PI案例

bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master yarn \
./examples/jars/spark-examples_2.11-2.1.1.jar \
10

—master yarn，表示Yarn方式运行

如果遇到 WARN Utils: Service ‘SparkUI’ could not bind on port 4040. Attempting port 4041.
java.lang.NoClassDefFoundError: com/sun/jersey/api/client/config/ClientConfig 报错 上面配置yarn-site.xml已经解决此报错问题

1.找到yarn下面相关包

find /usr/hdp/ |grep jersey

2.拷贝jar到spark

所缺的类在 jersey-core-1.9.jar 和 jersey-client-1.9.jar 两个jar包中 将 jersey-core-1.9.jar 和 jersey-client-1.9.jar 这两个包拷贝到$SPARK_HOME/jars目录下

4.1. 配置历史服务

由于是重新解压的Spark压缩文件，所以需要针对Yarn模式，再次配置一下历史服务器。

修改spark-default.conf.template

cd /opt/module/spark-yarn/conf/
mv spark-defaults.conf.template spark-defaults.conf
vi spark-defaults.conf

#配置spark历史服务
spark.eventLog.enabled          true
spark.eventLog.dir               hdfs://hadoop102:8020/directory

xsync spark-defaults.conf

修改 spark-env.sh配置

vi spark-env.sh

export SPARK_HISTORY_OPTS="
-Dspark.history.ui.port=18080 
-Dspark.history.fs.logDirectory=hdfs://hadoop102:8020/directory 
-Dspark.history.retainedApplications=30"

xsync spark-env.sh

5.2. 配置查看历史日志

为了从Yarn上关联到Spark历史服务器，需要配置关联路径

修改配置文件/opt/module/spark/conf/spark-defaults.conf

vim /opt/module/spark-yarn/conf/spark-defaults.conf

spark.yarn.historyServer.address=hadoop102:18080
spark.history.ui.port=18080

xsync /opt/module/spark-yarn/conf/spark-defaults.conf

启动spark历史服务

cd /opt/module/spark-yarn/
sbin/start-history-server.sh

重新提交任务

bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master yarn \
./examples/jars/spark-examples_2.11-2.1.1.jar \
10

查询http://hadoop103:8088/cluster

6. 运行流程

Spark有yarn-client和yarn-cluster两种模式，主要区别在于：Driver程序的运行节点。

yarn-client：Driver程序运行在客户端，适用于交互、调试，希望立即看到app的输出。

yarn-cluster：Driver程序运行在由ResourceManager启动的APPMaster适用于生产环境。

客户端模式（默认）

bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master yarn \
--deploy-mode client \
./examples/jars/spark-examples_2.11-2.1.1.jar \
10

集群模式

bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master yarn \
--deploy-mode cluster \
./examples/jars/spark-examples_2.11-2.1.1.jar \
10

默认无法访问需要在yarn-site.xml添加配置并启动yarn历史服务器

vim /opt/module/hadoop-3.1.3/etc/hadoop/yarn-site.xml

<property>
    <name>yarn.log.server.url</name>
    <value>http://hadoop104:19888/jobhistory/logs</value>
</property>

启动历史服务器

mapred --daemon start historyserver

http://hadoop102:19888/jobhistory/logs/hadoop103:44236/container_1639655468064_0005_01_000001/container_1639655468064_0005_01_000001/atguigu/stdout?start=-4096

7. Mesos模式

Spark客户端直接连接Mesos；不需要额外构建Spark集群。国内应用比较少，更多的是运用Yarn调度。

8. 几种模式对比

9. 端口号总结

1. Spark历史服务器端口号：18080 （类比于Hadoop历史服务器端口号：19888）
2. Spark Master Web端口号：8080（类比于Hadoop的NameNode Web端口号：9870(50070)）
3. Spark Master内部通信服务端口号：7077 （类比于Hadoop的8020(9000)端口）
4. Spark查看当前Spark-shell运行任务情况端口号：4040
5. Hadoop YARN任务运行情况查看端口号：8088

10. WordCount案例

Spark Shell仅在测试和验证我们的程序时使用的较多，在生产环境中，通常会在IDE中编制程序，然后打成Jar包，然后提交到集群，最常用的是创建一个Maven项目，利用Maven来管理Jar包的依赖。

新建maven项目 并添加scala框架支持 导入pom文件

<dependencies>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-core_2.11</artifactId>
        <version>2.1.1</version>
    </dependency>
</dependencies>
<build>
    <finalName>WordCount</finalName>
    <plugins>
        <plugin>
            <groupId>net.alchim31.maven</groupId>
            <artifactId>scala-maven-plugin</artifactId>
            <version>4.5.3</version>
            <executions>
                <execution>
                   <goals>
                      <goal>compile</goal>
                      <goal>testCompile</goal>
                   </goals>
                </execution>
             </executions>
        </plugin>
    </plugins>
</build>

如果maven版本为3.2.x，插件下载报错，那么修改插件版本为3.3.2

创建伴生对象WordCount，编写代码

package com.atguigu.spark.day01

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object WordCount {
  def main(args: Array[String]): Unit = {
    //创建SparckConfig配置文件
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    //创建sparkContext对象
    val sc: SparkContext = new SparkContext(conf)
    //读取外部数据
    val textRDD: RDD[String] = sc.textFile(args(0))
    //对读取到的内容进行切割并进行扁平化操作
    val flatMapRDD: RDD[String] = textRDD.flatMap(_.split(" "))
    //对数据集中的内容进行结构的转换 -- 计数
    val mapRDD: RDD[(String, Int)] = flatMapRDD.map((_, 1))
    //对相同单词 出现次数进行汇总
    val reduceRDD: RDD[(String, Int)] = mapRDD.reduceByKey(_ + _)
    //存储为文件
    reduceRDD.saveAsTextFile(args(1))
//    val res: Array[(String, Int)] = reduceRDD.collect()

    //释放资源
    sc.stop()
  }

}

本地运行

添加打包插件

     <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-assembly-plugin</artifactId>
                <version>3.0.0</version>
                <configuration>
                    <archive>
                        <manifest>
                            <mainClass>com.atguigu.spark.day01.WordCount</mainClass>
                        </manifest>
                    </archive>
                    <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>

maven点击package打包，将WordCount.jar(不带依赖)上传到/opt/module/spark-yarn目录

在HDFS上创建，存储输入数据文件的路径/input

hadoop fs -mkdir /input
hadoop fs -put /opt/module/spark-local-standalone/input/1.txt /input

执行任务

cd /opt/module/spark-yarn
bin/spark-submit \
--class com.atguigu.spark.day01.WordCount \
--master yarn \
WordCount.jar \
/input \
/output