Spark环境的安装

---

一、 Spark简介

1.1 Spark是什么

是一种基于内存的快速、通用、可拓展的大数据分析计算引擎。

1.2 Hadoop 和 Spark关联

1. hadoop ：2013年10月发布2.X (Yarn)版本；
2. spark ： 2013年6月，Spark成为了Apache基金会下的项目。
3. Spark可以理解为hadoop MR的升级版。

1.2.1 hadoop发展历史

-- 1.X 版本 --2011年发布

从架构的角度存在很多的问题
1. Namenode 是单点操作，所以容易出现单点故障，制约了HDFS的发展

2. Namenode的内存限制也影响了HDFS的发展
 
3. MapReduce是一种基于数据集的工作模式，面向数据，这种工作模式一般是从存储上加载数据集，然后操作数据集，最后将结果写入物理存储设备。数据更多面临的是一次性计算，所以初衷是单一数据计算，不支持迭代计算
 
4. 资源调度和任务调度耦合在一起，无法扩展，所以Hadoop1.X版本只支持MR计算框架


-- 2.X 版本（Yarn） --2013.10月发布
 
1. 2.X版本支持Namenode高可用
 
2. 2.X版本使用新的资源调度框架Yarn，只做资源调度，不进行任务调度。
 
3. MR框架只做任务调度，可插拔，所以扩展性非常的强


1.2.2 Spark 技术

1. Spark其实核心思想就是基于MR的，优化了MR数据处理的中间过程，提升了数据处理的性能

2. MR：多任务之间的数据会进行落盘

3. Spark：多任务之间的数据会在内存中。

4. 因为内存大小也有上限的，当内存不足时，就会出现job运行失败，所以Spark并不是完全替代MR。

• MR和Spark区别



• Spark的特点

1. spark计算模型更加丰富，MR只有mapper和reducer， spark的计算模型模糊了mapper和reduce的界限，更容易使用;
2. spark使用scala语言开发，支持函数式编程，所以就更利用迭代式计算.
3. spark也有自己的任务调度器和资源调度器。


1.2.3 Spark On Yarn

在实际开发中，hadoop和Spark合二为一。

调度器：Hadoop的Yarn

任务执行：Spark的任务调度，Driver和Executor


1.3 Spark的核心框架


1. Apache Spark Core : 最基础和最核心的功能
2. Spark SQL :用于处理关系型数据库
3. Spark Streaming：针对实时数据的处理流式计算的框架，Flink框架更有优势
4. Spark MLlib: 机器学习
5. Spark Graphx:面向图形计算

我们重点学习Spark前面三个框架。

二、Spark快速上手

创建Maven工程

2.1 增加Scala插件

--当前使用的Spark版本为2.4.5，默认采用的Scala版本为2.12

2.2 增加依赖关系

--修改Maven项目中的POM文件，增加Spark框架的依赖关系。当前文件是基于Spark2.4.5版本，使用时请注意对应版本

<dependencies>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-core_2.12</artifactId>
        <version>2.4.5</version>
    </dependency>
</dependencies>
<build>
    <plugins>
        <!-- 该插件用于将Scala代码编译成class文件 -->
        <plugin>
            <groupId>net.alchim31.maven</groupId>
            <artifactId>scala-maven-plugin</artifactId>
            <version>3.2.2</version>
            <executions>
                <execution>
                    <!-- 声明绑定到maven的compile阶段 -->
                    <goals>
                        <goal>testCompile</goal>
                    </goals>
                </execution>
            </executions>
        </plugin>
        <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>
    </plugins>
</build>

2.3 增加日志文件配置文件

-- 在项目的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

2.4 WordCount

• 方法1：

步骤：
//1. 创建Spark的环境
//2. 连接Spark
//3. 具体的操作
//4. 关闭连接

object Spark_WordCount {
  def main(args: Array[String]): Unit = {
    //需求：读取一个文件中的数据，求(word,count)

    //1. 创建Spark的环境
    val sparkConf: SparkConf = new SparkConf().setMaster("local").setAppName("wordcount")

    //2. 连接Spark
    val sc = new SparkContext(sparkConf)

    //3. 具体的操作
    //3.1 读取数据
    val str: RDD[String] = sc.textFile("input")

    //3.2 扁平化数据
    val words: RDD[String] = str.flatMap(_.split(" "))

    //3.3 分组
    val wordtocount: RDD[(String, Iterable[String])] = words.groupBy(word => word)

    //3.4 结构化处理
    val wordcount: RDD[(String, Int)] = wordtocount.map {
      case (word, iter) => (word, iter.size)
    }

    //3.5 数据采集并打印在控制台
    val result: Array[(String, Int)] = wordcount.collect()
    println(result.mkString(","))

    //4. 关闭连接
    sc.stop()
    

  }

}

• 方法二

-- 在方法1的基础上，将分组和结构化处理操作进行优化。
-- 方法1：//3.3 分组 +  //3.4 结构化处理
分组操作结果：(spark , List(spark,spark)) 
结构化处理结果：(spark,list.length)

-- 优化方法：在进行分组操作的同时就进行单词出现的次数进行计算。

object Spark_WordCount1 {
  def main(args: Array[String]): Unit = {

    //1. 创建spark的环境
    val sparkConf: SparkConf = new SparkConf().setMaster("local").setAppName("wordcount")

    //2. 与spark进行连接
    val sc = new SparkContext(sparkConf)

    //3. 具体的操作

    val datas: RDD[String] = sc.textFile("input")

    val words: RDD[String] = datas.flatMap(_.split(" "))

    val wordto: RDD[(String, Int)] = words.map((_,1))

    val wordcount: RDD[(String, Int)] = wordto.reduceByKey(_ + _)

    println(wordcount.collect().mkString(","))
    //4. 关闭连接
    sc.stop()
    

  }

}

三、 Spark运行环境

3.1 Spark的运行环境


我们主要学习3种环境，并简单讲述一下2种环境

学习的环境：
1. 本地运行模式：local
2. 独立运行模式：standalone
3. YARN运行模式

简单了解：
1. MESOS
2. Windows

3.2 本地运行模式

3.2.1 本地模式介绍

-- 1. 什么是本地模式
 不需要其他任何节点资源就可以在本地执行Spark代码的环境.
 我们之前在IDEA中运行的环境我们称之为开发环境，和本地环境还是不一样

3.2.2 解压缩文件

• 第一步：解压缩

1. 将spark-2.4.5-bin-without-hadoop-scala-2.12.tgz文件上传到Linux中/opt/software，并解压缩到/opt/module文件目录下，并修改名称为：spark-local

//解压
tar -zxvf spark-2.4.5-bin-without-hadoop-scala-2.12.tgz -C /opt/module
//改名
mv spark-2.4.5-bin-without-hadoop-scala-2.12 spark-local

• 第二步：Spark关联hadoop，spark2.4.5默认不支持Hadoop3，可以采用多种不同的方式关联Hadoop3

1. 方法1：修改spark-local/conf/spark-env.sh文件，增加如下内容

SPARK_DIST_CLASSPATH=$(/opt/module/hadoop3/bin/hadoop classpath)

2. 方法2：将hadoop3的jar包上传到==/opt/module/spark-local/jars==

3.2.3 启动local环境

1. 进入/opt/module/spark-local路径，执行如下指令

[atguigu@hadoop105 spark-local]$ bin/spark-shell --master local


2. 启动成功后，可以输入网址进行Web UI监控页面访问

http://hadoop105:4040


3.2.4 命令行工具

1. 在==/opt/module/spark-local/data==目录下创建一个文件：word.txt，在文件中随机写入一些单词
2. 在==/opt/module/spark-local==下，执行如下命令

[atguigu@hadoop105 spark-local]$ sc.textFile("data/word.txt").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).collect


3.2.4 退出本地模式

• 按键Ctrl+C或输入Scala指令

：quit

3.2.5 提交应用

[atguigu@hadoop105 spark-local]$ bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master local[2] \
./examples/jars/spark-examples_2.12-2.4.5.jar \
10

• 参数说明

spark-submit ： 提交应用程序
--class 表示要执行程序的主类
--master local[2] ： 本地部署模式，local，本地，[2],数字2表示分配的cpu核数
spark-examples_2.12-2.4.5.jar ： 执行程序的主类所在的jar包
10 ： 用于设定当前应用的任务数量


3.3 独立运行模式

3.3.1 独立运行介绍

-- 什么是独立运行模式
   1. standalone，表示资源调度器和任务调度均是使用Sqark自身的集群来运行。
      a、资源调度：master(Spark的调度者),worker(Spark节点)
      b、任务调度：driver(驱动器)，exerutor(执行器)

3.3.2 解压缩文件

• 第一步：解压缩

1. 将spark-2.4.5-bin-without-hadoop-scala-2.12.tgz文件上传到Linux中/opt/software，并解压缩到/opt/module文件目录下，并修改名称为：spark-standalone

//解压
tar -zxvf spark-2.4.5-bin-without-hadoop-scala-2.12.tgz -C /opt/module
//改名
mv spark-2.4.5-bin-without-hadoop-scala-2.12 spark-standalone

• 第二步：Spark关联hadoop，spark2.4.5默认不支持Hadoop3，可以采用多种不同的方式关联Hadoop3

1. 方法1：修改spark-local/conf/spark-env.sh文件，增加如下内容

SPARK_DIST_CLASSPATH=$(/opt/module/hadoop3/bin/hadoop classpath)

2. 方法2：将hadoop3的jar包上传到==/opt/module/spark-local/jars==

3.3.3 修改配置文件

1. 进入/opt/module/spark-standalone/conf，修改slaves.template文件名为slaves

[atguigu@hadoop105 conf]$ mv slaves.template slaves

2. 修改slaves文件，添加work节点

hadoop105
hadoop106
hadoop107

3. 修改spark-env.sh.template文件名为spark-env.sh

[atguigu@hadoop105 conf]$ mv spark-env.sh.template spark-env.sh

4. 修改spark-env.sh文件，添加JAVA_HOME环境变量和集群对应的master节点

export JAVA_HOME=/opt/module/jdk1.8.0_212
SPARK_MASTER_HOST=hadoop105
SPARK_MASTER_PORT=7077

注意：7077端口，相当于hadoop3内部通信的8020端口

5. 分发spark-standalone目录

xsync spark-standalone

3.3.4 启动集群

1. 执行脚本命令：

-- 打开集群,先启动Master，再启动worker
[atguigu@hadoop105 spark-standalone]$ sbin/start-all.sh
-- 关闭集群,先关闭worker，再关闭Master
[atguigu@hadoop105 spark-standalone]$ sbin/stop-all.sh

2. 查看三台服务器的进程

[atguigu@hadoop105 spark-standalone]$ myjps


3. 查看Master的资源监控Web UI 网页界面：http://hadoop105:8080


3.3.5 提交应用

[atguigu@hadoop105 spark-standalone]$bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master spark://hadoop105:7077 \
./examples/jars/spark-examples_2.12-2.4.5.jar \
10

• 参数说明

spark-submit  -- 提交应用
--class       -- 表示要执行程序的主类
--master spark://hadoop105:7077 -- 独立运行模式，7070为spark内部通信的端口
spark-examples_2.12-2.4.5.jar   -- 程序主类所在的jar
10  -- 设定当前应用的任务数量

3.3.6 提交参数说明

bin/spark-submit \
--class <main-class>
--master <master-url> \
... # other options
<application-jar> \
[application-arguments]

参数	解释[可选值举例 ]
--class	Spark程序中包含主函数的类
--master	Spark程序运行的模式 [本地模式：local[*]、spark://linux1:7077、Yarn ]
--executor-memory 1G	指定每个executor可用内存为1G [符合集群内存配置即可，具体情况具体分析 ]
--total-executor-cores 2	指定所有executor使用的cpu核数为2个
--executor-cores	指定每个executor使用的cpu核数
application-jar	打包好的应用jar，包含依赖。这个URL在集群中全局可见。 比如hdfs:// 共享存储系统，如果是file:// path，那么所有的节点的path都包含同样的jar
application-arguments	传给main()方法的参数

3.3.7 配置历史服务器

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

1. 修改spark-defaults.conf.template文件名为spark-defaults.conf

[atguigu@hadoop105 conf]$ mv spark-defaults.conf.template spark-defaults.conf

2. 修改spark-default.conf文件，配置日志存储路径

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

注意：需要启动hadoop集群，HDFS上的directory目录需要提前存在

3. 修改spark-env.sh文件, 添加日志配置

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

• 参数1含义：WEBUI访问的端口号为18080

• 参数2含义：指定历史服务器日志存储路径

• 参数3含义：指定保存Application历史记录的个数，如果超过这个值，旧的应用程序信息将被删除，这个是内存中的应用数，而不是页面上显示的应用数。

4. 分发配置文件

xsync conf

5. 重新启动集群和历史服务

[atguigu@hadoop105 spark-standalone]$ sbin/start-all.sh 
[atguigu@hadoop105 spark-standalone]$ sbin/start-history-server.sh

6. 重新执行任务

[atguigu@hadoop105 spark-standalone]$bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master spark://hadoop105:7077 \
./examples/jars/spark-examples_2.12-2.4.5.jar \
10

7. 查看历史服务：http://hadoop105:18080


3.3.8 配置高可用

-- 因为Spark集群中的Master只有一个，存在单点故障，所以需要给集群配置多个Master，一旦处于活跃的Master故障时，StandBy的Master转换为活跃的，提供服务。

• 集群规划

	Hadoop105	Hadoop106	Hadoop107
Spark	MasterZookeeperWorker	MasterZookeeperWorker	ZookeeperWorker

1. 停止集群

[atguigu@hadoop105 spark-standalone]$ sbin/stop-all.sh

2. 启动Zookeeper

[atguigu@hadoop105 spark-standalone]$ zk start

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

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

SPARK_MASTER_WEBUI_PORT=8989

#添加如下内容:
export SPARK_DAEMON_JAVA_OPTS="
-Dspark.deploy.recoveryMode=ZOOKEEPER 
-Dspark.deploy.zookeeper.url=hadoop105,hadoop106,hadoop107 
-Dspark.deploy.zookeeper.dir=/spark"

4. 分发配置文件

[atguigu@hadoop105 spark-standalone]$ xsync conf/

5. 启动集群

[atguigu@hadoop105 spark-standalone]$ sbin/start-all.sh

6. 启动lhadoop106的单独Master节点，此时hadoop106节点Master状态处于备用状态

[atguigu@hadoop106 spark-standalone]$ sbin/start-master.sh


7. 停止hadoop105的Master资源监控进程
8. 查看hadoop106的Master 资源监控Web UI，稍等一段时间后，hadoop106节点的Master状态提升为活动状态


3.4 YARN

1. Spark重点在与实时的计算引擎，而不是资源框架，所以本身提供资源调度并不是其本身。
2. 所以参考在YARN环境下运行Spark。
3. 资源调度：hadoop的yarn
   执行任务： spark的driver和executor

3.4.1 解压缩文件

• 第一步：解压缩

1. 将spark-2.4.5-bin-without-hadoop-scala-2.12.tgz文件上传到Linux中/opt/software，并解压缩到/opt/module文件目录下，并修改名称为：spark-yarn

//解压
tar -zxvf spark-2.4.5-bin-without-hadoop-scala-2.12.tgz -C /opt/module
//改名
mv spark-2.4.5-bin-without-hadoop-scala-2.12 spark-yarn

• 第二步：Spark关联hadoop，spark2.4.5默认不支持Hadoop3，可以采用多种不同的方式关联Hadoop3

1. 方法1：修改spark-local/conf/spark-env.sh文件，增加如下内容

SPARK_DIST_CLASSPATH=$(/opt/module/hadoop3/bin/hadoop classpath)

2. 方法2：将hadoop3的jar包上传到==/opt/module/spark-local/jars==

3.4.2 修改配置文件

1. 修改hadoop配置文件/opt/module/hadoop/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>

2. 修改conf/spark-env.sh，添加JAVA_HOME和YARN_CONF_DIR配置

-- 改名
[atguigu@hadoop105 conf]$ mv spark-env.sh.template spark-env.sh

-- 添加配置
[atguigu@hadoop105 conf]$ vim spark-env.sh

export JAVA_HOME=/opt/module/jdk1.8.0_212
YARN_CONF_DIR=/opt/module/hadoop-3.1.3/etc/hadoop

3.4.3 启动hadoop集群

[atguigu@hadoop105 conf]$ mycluster start

3.4.4 提交应用

[atguigu@hadoop105 spark-standalone]$bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master spark://hadoop105:7077 \
./examples/jars/spark-examples_2.12-2.4.5.jar \
10

查看http://hadoop105:8088页面，点击History，查看历史页面

3.4.5 配置历史服务器

1. 修改spark-defaults.conf.template文件名为spark-defaults.conf

[atguigu@hadoop105 conf]$ mv spark-defaults.conf.template spark-defaults.conf

2. 修改spark-default.conf文件，配置日志存储路径

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

注意：需要启动hadoop集群，HDFS上的directory目录需要提前存在

3. 修改spark-env.sh文件, 添加日志配置

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

• 参数1含义：WEBUI访问的端口号为18080

• 参数2含义：指定历史服务器日志存储路径

• 参数3含义：指定保存Application历史记录的个数，如果超过这个值，旧的应用程序信息将被删除，这个是内存中的应用数，而不是页面上显示的应用数。

4. 修改spark-defaults.conf

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

5. 启动历史服务

sbin/start-history-server.sh

6. 重新提交应用

[atguigu@hadoop105 spark-standalone]$bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master spark://hadoop105:7077 \
./examples/jars/spark-examples_2.12-2.4.5.jar \
10


7. Web页面查看日志：http://hadoop106:8088

3.5 K8S & Mesos模式

Mesos是Apache下的开源分布式资源管理框架，它被称为是分布式系统的内核,在Twitter得到广泛使用,管理着Twitter超过30,0000台服务器上的应用部署，但是在国内，依然使用着传统的Hadoop大数据框架，所以国内使用Mesos框架的并不多，但是原理其实都差不多，这里我们就不做过多讲解了。


容器化部署是目前业界很流行的一项技术，基于Docker镜像运行能够让用户更加方便地对应用进行管理和运维。容器管理工具中最为流行的就是Kubernetes（k8s），而Spark也在最近的版本中支持了k8s部署模式


3.6 Windows 模式

1. 将文件spark-2.4.5-bin-without-hadoop-scala-2.12.tgz解压缩到无中文无空格的路径中，将hadoop3依赖jar包拷贝到jars目录中
2. 执行解压缩文件路径下bin目录中的spark-shell.cmd文件，启动Spark本地环境
3. 在bin目录中创建input目录，并添加word.txt文件, 在命令行中输入脚本代码

sc.textFile("input").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).collect

3.7 部署模式的对比

模式	Spark安装机器数	需启动的进程	所属者	应用场景
Local	1	无	Spark	测试
Standalone	3	Master及Worker	Spark	单独部署
Yarn	1	Yarn及HDFS	Hadoop	混合部署

3.8 端口号

端口	作用
8080	资源的监控页面端口
7077	Spark的worker内部通信的端口
4040	计算的监控页面端口
18080	历史服务器端口
8088	RM的资源监控端口