1. Spark概述

1.1 什么是Spark

Spark是一种基于内存的快速、通用、可扩展的大数据分析框架。

1.2 Hadoop和Spark

• Hadoop：一次性计算框架，基于磁盘，不适合迭代式计算。框架在处理数据的时候，会冲存储设备将数据读取出来，进行逻辑处理，然后将处理结果重新存储到介质中。

• Spark：基于内存，适合迭代式计算，任务启动快，但内存会有限制。

1.3 Spark核心模块

• Spark Core：提供了丰富的API，供其他模块使用。
• Spark Streaming：Spark提供的流式计算组件。
• Spark SQL： 用于调用HIve和关系型数据库的组件。
• Spark MLlib：机器学习
• Spark GraphX：面向图形计算提供的框架和算法库。

2. 快速上手

自行安装环境

• Scala：2.12.11
• Spark：3.0.0

2.1 项目目录结构

spark-learning-- spark-core

2.2 实现WordCount三种方式

• 第一种

import org.apache.spark.{SparkConf, SparkContext}object Spark01_WordCount {def main(args: Array[String]): Unit = {// 1. 创建连接val sparkConf = new SparkConf().setMaster("local").setAppName("WordCount")val sc = new SparkContext(sparkConf)// 2. 执行业务逻辑val lines = sc.textFile("datas")// 将单词扁平化val words = lines.flatMap(_.split(" "))// 将单词进行分组val wordGroup = words.groupBy(word => word)val value = wordGroup.map {case (word, list) =>(word, list.size)}// 打印value.foreach(println)// 3. 关闭连接sc.stop}}

• 第二种

import org.apache.spark.{SparkConf, SparkContext}object Spark02_WordCount {def main(args: Array[String]): Unit = {val sparkConf = new SparkConf().setMaster("local").setAppName("WordCount")val sc = new SparkContext(sparkConf)val lines = sc.textFile("datas")val words = lines.flatMap(_.split(" "))val wordToOne = words.map(word => (word, 1))val wordGroup = wordToOne.groupBy(t => t._1)val value = wordGroup.map {case (word, list) =>list.reduce((t1, t2) => {(t1._1, t2._2 + t1._2)})}// 打印value.foreach(println)// 3. 关闭连接sc.stop}}

• 第三种

import org.apache.spark.{SparkConf, SparkContext}object Spark03_WordCount {def main(args: Array[String]): Unit = {// 1. 创建连接val sparkConf = new SparkConf().setMaster("local").setAppName("WordCount")val sc = new SparkContext(sparkConf)// 2. 执行业务逻辑val lines = sc.textFile("datas")// 将单词扁平化val words = lines.flatMap(_.split(" "))// 将单词进行分组val wordGroup = words.map(word => (word, 1))val value = wordGroup.reduceByKey(_ + _)// 打印value.foreach(println)// 3. 关闭连接sc.stop}}

3. Spark运行环境

由于之前有配置Hadoop环境， 这里就不再赘述Hadoop的相关配置了。

3.1 Local环境

不需要借助任何其他节点就可以运行的环境就是本地环境，单独在idea运行的是开发环境。

3.1.1 环境安装

# 解压
[hadoop@hadoop102 module]$ tar -zxvf spark-3.0.0-bin-hadoop3.2.tgz -C /opt/module/
# 重命名
[hadoop@hadoop102 module]$ cd /opt/module/
[hadoop@hadoop102 module]$ mv spark-3.0.0-bin-hadoop3.2/ spark-local

3.1.2 测试是否安装成功

出现下面这个界面表示安装成功。

[hadoop@hadoop102 spark-local]$ bin/spark-shell 
21/10/18 13:44:29 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
Spark context Web UI available at http://hadoop102:4040
Spark context available as 'sc' (master = local[*], app id = local-1634535874212).
Spark session available as 'spark'.
Welcome to____              __/ __/__  ___ _____/ /___\ \/ _ \/ _ `/ __/  '_//___/ .__/\_,_/_/ /_/\_\   version 3.0.0/_/Using Scala version 2.12.10 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_212)
Type in expressions to have them evaluated.
Type :help for more information.scala> 

3.1.3 提交作业

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

3.2 Standalone模式

独立部署模式。体现了经典的主从模式。

• 集群规划

	hadoop102	hadoop103	Hadoop104
Spark	Worker、Matser	Worker	Worker

3.2.1 环境安装

# 解压
[hadoop@hadoop102 spark]$ tar -zxvf spark-3.0.0-bin-hadoop3.2.tgz -C /opt/module/
# 重命名
[hadoop@hadoop102 module]$ cd /opt/module/
[hadoop@hadoop102 module]$ mv spark-3.0.0-bin-hadoop3.2/ spark-standalone

3.2.2 配置

# 重命名
[hadoop@hadoop102 conf]$ mv slaves.template slaves
# 配置slaves
[hadoop@hadoop102 conf]$ vim slaves 
hadoop102
hadoop103
hadoop104# 重命名
[hadoop@hadoop102 conf]$ mv spark-env.sh.template spark-env.sh# 添加配置
export JAVA_HOME=/opt/module/jdk1.8.0_212
SPARK_MASTER_HOST=hadoop102
SPARK_MASTER_PORT=7077

3.2.3 分发

[hadoop@hadoop102 conf]$ xsync /opt/module/spark-standalone/

3.2.4 启动服务

这里有个问题就是我之前安装的spark程序配置了环境变量，因此在启动服务的时候就只能启动原来目录下的spark程序，看了日志之后发现spark_home不对，需注释掉原来的配置就可以了。

[hadoop@hadoop102 spark-standalone]$ sbin/start-all.sh 
starting org.apache.spark.deploy.master.Master, logging to /opt/module/spark-standalone/logs/spark-hadoop-org.apache.spark.deploy.master.Master-1-hadoop102.out
hadoop102: starting org.apache.spark.deploy.worker.Worker, logging to /opt/module/spark-standalone/logs/spark-hadoop-org.apache.spark.deploy.worker.Worker-1-hadoop102.out
hadoop104: starting org.apache.spark.deploy.worker.Worker, logging to /opt/module/spark-standalone/logs/spark-hadoop-org.apache.spark.deploy.worker.Worker-1-hadoop104.out
hadoop103: starting org.apache.spark.deploy.worker.Worker, logging to /opt/module/spark-standalone/logs/spark-hadoop-org.apache.spark.deploy.worker.Worker-1-hadoop103.out
[hadoop@hadoop102 spark-standalone]$ xcall.sh  jps
============hadoop102================
1700 Jps
1560 Master
1644 Worker
============hadoop103================
1249 Worker
1301 Jps
============hadoop104================
1249 Jps
1195 Worker

3.2.5 查看web页面

http://hadoop102:8080/

3.2.6 提交作业

bin/spark-submit --class org.apache.spark.examples.SparkPi --master spark://hadoop102:7077 ./examples/jars/spark-examples_2.12-3.0.0.jar 10

• class：Spark程序中包含主程序的类
• master：Spark 用来指定执行模式 local、master（spark://hadoop102:7077）、yarn
• Executor-memory 1G：指定每个executor可用内存
• Total-executors-core 2：指定所有executor可用CPU核数
• application-jar：打包好的应用jar，此参数的名字可以不写
• application- arguments：传递给main方法的参数

3.2.7 配置历史服务

# 重命名配置文件
[hadoop@hadoop102 conf]$ mv spark-defaults.conf.template spark-defaults.conf
# 添加内容
spark.eventLog.enabled           true
spark.eventLog.dir               hdfs://hadoop102:8020/spark-log# 启动hdfs
[hadoop@hadoop102 conf]$ start-dfs.sh 
# 创建目录
[hadoop@hadoop102 conf]$ hadoop fs -mkdir /spark-log# 修改spark-env.sh
[hadoop@hadoop102 spark-standalone]$ vim conf/spark-env.sh
# history server
export SPARK_HISTORY_OPTS="
-Dspark.history.ui.port=18080
-Dspark.history.fs.logDirectory=hdfs://hadoop102:8020/spark-log
-Dspark.history.retainedApplications=30"# 重新启动服务
[hadoop@hadoop102 spark-standalone]$ sbin/stop-all.sh 
[hadoop@hadoop102 spark-standalone]$ sbin/start-all.sh 
[hadoop@hadoop102 spark-standalone]$ sbin/start-history-server.sh# 访问web历史服务
http://hadoop102:18080/

3.2.8 配置高可用（HA）

# 修改配置文件
[hadoop@hadoop102 spark-standalone]$ vim conf/spark-env.sh 
# 注释掉下面两行
# SPARK_MASTER_HOST=hadoop102
# SPARK_MASTER_PORT=7077# 增加下面配置
export SPARK_MASTER_WEBUI_PORT=8899
export SPARK_DAEMON_JAVA_OPTS="
-Dspark.deploy.recoveryMode=ZOOKEEPER
-Dspark.deploy.zookeeper.url=hadoop102,hadoop103,hadoop104
-Dspark.deploy.zookeeper.dir=/spark"
# 分发
[hadoop@hadoop102 spark-standalone]$ xsync conf/# 启动zk
[hadoop@hadoop102 spark-standalone]$ zk.sh start# 重启服务
[hadoop@hadoop102 spark-standalone]$ sbin/stop-all.sh
[hadoop@hadoop102 spark-standalone]$ sbin/start-all.sh 
[hadoop@hadoop102 spark-standalone]$ sbin/stop-history-server.sh 
[hadoop@hadoop102 spark-standalone]$ sbin/start-history-server.sh 
# hadoop103 启动master
[hadoop@hadoop103 spark-standalone]$ sbin/start-master.sh# 查看两个master服务状态，一个active、一个standby
http://hadoop103:8899/
http://hadoop102:8899/

3.3 Yarn模式

Spark框架在Standalone模式下自身提供计算资源，无需与其他框架整合，降低了耦合性，但是Spark是计算框架，不是资源调度框架，所以提供资源调度并不是他的强项，所以还是需要跟其他框架整合比较好一些。国内更多的使用yarn模式，国外的更多使用mesos。

3.3.1 安装

# 解压
[hadoop@hadoop102 spark]$ tar -zxvf spark-3.0.0-bin-hadoop3.2.tgz -C /opt/module/
# 重命名
[hadoop@hadoop102 spark]$ cd /opt/module/
[hadoop@hadoop102 module]$ mv spark-3.0.0-bin-hadoop3.2/ spark-yarn

3.3.2 配置

# 重命名
[hadoop@hadoop102 conf]$ mv spark-env.sh.template spark-env.sh# 添加内容
export JAVA_HOME=/opt/module/jdk1.8.0_212
export HADOOP_CONF_DIR=/opt/module/hadoop-3.1.3/etc/hadoop

3.3.3 启动服务

这里不需要去启动服务器，只需要启动Hdfs、Yarn即可。

[hadoop@hadoop102 spark-yarn]$ sbin/start-all.sh

3.3.4 提交任务

[hadoop@hadoop102 spark-yarn]$ bin/spark-submit --class org.apache.spark.examples.SparkPi --master yarn --deploy-mode cluster ./examples/jars/spark-examples_2.12-3.0.0.jar 10

3.3.5 配置历史服务

# 修改配置文件
[hadoop@hadoop102 conf]$ mv spark-defaults.conf.template spark-defaults.conf
# 添加内容
spark.eventLog.enabled           true
spark.eventLog.dir               hdfs://hadoop102:8020/spark-log# 修改spark-env.sh
[hadoop@hadoop102 conf]$ vim spark-env.sh 
# history server
export SPARK_HISTORY_OPTS="
-Dspark.history.ui.port=18080
-Dspark.history.fs.logDirectory=hdfs://hadoop102:8020/spark-log
-Dspark.history.retainedApplications=30"# 重新提交任务
bin/spark-submit --class org.apache.spark.examples.SparkPi --master yarn --deploy-mode client ./examples/jars/spark-examples_2.12-3.0.0.jar 10# 查看历史服务器
http://hadoop102:18080/
# 查看yarn任务提交
http://hadoop103:8088/cluster

3.3.6 Yarn的两种模式

• Client：用于测试，由于Driver运行在本地客户端，负责调度application，与yarn集群交互会产生大量的网络通信，从而导致网卡流量激增。好处在于可以看到日志，便于调试。
• Cluster：用于生产，因为Driver运行在NodeManager，没有网卡流量激增的情况，但由于日志都在服务器上，所以查看日志比较困难，需要到服务器上去看。

所以在服务器上只需要开启一种模式即可，不需要开启两种，会出现资源竞争的问题。

3.4 三种部署模式对比

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

3.5 端口号

• 4040：查看Spark-shell执行任务情况
• 7077：Spark Master内部通信地址
• 8080：Standalone模式下Spark Master Web资源查看页面
• 18080：历史服务器端口
• 8088：Yarn任务运行情况查询端口

4. Spark核心概念

Spark的核心是一个计算引擎，从整体上来讲，他采用了matser-slave的结构，是一个标准的主从结构。

其中driver代表了master，负责整个集群的作业任务调度。executor代表slave，负责任务执行。

4.2 Driver

Spark驱动器节点，用于执行Spark任务main方法，负责实际代码的执行工作。Dirver在Spark作业中的主要职责：

• 将用户程序转化为作业（JOB）
• 在Executor之间调度任务（Task）
• 跟踪Executor执行情况
• 通过UI展示任务运行情况

4.3 Executor

核心功能：

• 负责运行组成Spark应用的任务，并将结果返回给驱动器
• 他们通过自身的BlockManager为用户提供RDD运行的内存。RDD是直接缓存在Executor进行内的，因此任务可以在运行时加速运算。

4.4 Master&Worker

• Master：主要负责资源调度和分配，监控集群，类似于Yarn中ResourceManager
• Worker：类似于Yarn中NodeManager

4.5 ApplicationMatser

作为Driver和Master中间的桥梁。

5. Spark核心编程

Spark计算框架为了能够进行高并发和高吞吐量的数据处理，封装了三大数据结构，用于处理不同的应用场景。三大数据结构分别是：

• RDD：弹性分布式数据集
• 累加器：分布式共享只写变量
• 广播变量：分布式共享只读变量

RDD的处理模式类似于IO流也有装饰者模式。

RDD只有在调collect方法的时候才会执行业务逻辑，之前的操作都是一些封装。

RDD是不保存数据的，但IO是会保存一部分数据的。