大数据开发之Spark（入门）

第 1 章：Spark概述

1.1 什么是spark

回顾：hadoop主要解决，海量数据的存储和海量数据的分析计算。
spark是一种基于内存的快速、通用、可扩展的大数据分析计算引擎。

1.2 hadoop与spark历史

hadoop的yarn框架比spark框架诞生的晚，所以spark自己也涉及了一套资源调度框架。
区别：
1、mr是基于磁盘的，spark是基于内存
2、mr的task是进程
3、spark的task是线程，在executor进程里执行的是线程
4、mr在container里执行（留有接口方便插入），spark在worker里执行（自己用，没有接口）
5、mr适合做一次计算，spark适合做迭代计算

1.3 hadoop与spark框架对比

1、hadoop mr框架
从数据源获取数据，经过分析计算，将结果输出到指定位置，核心是一次计算，不适合迭代计算。

2、spark框架
spark框架计算比mr快的原因是：中间结果不落盘。注意spark的shuffle也是落盘的。

1.4 spark内置模块

spark core：实现了spark的基本功能，包括任务调度、内存管理、错误恢复、与存储系统交互等模块。spark core中还包含了对弹性分布式数据集（resilient distributed dataset，简称rdd）的api定义。
spark sql：是spark用来操作结构化数据的程序包。通过spark sql，我们可以使用sql或者apache hive版本的hql来查询数据。spark sql支持多种数据源，比如hive表、parquet以及json等。
spark mllib：提供常见的机器学习功能的程序库。包括分类、回归、聚类、协同过滤等，还提供了模型评估、数据导入等额外的支持功能。
spark graphx：主要用于图形并行计算和图挖掘系统的组件。
集群管理器：spark设计为可以高效地在一个计算节点到数千个计算节点之间伸缩计算。为了实现这样的要求，同时获得最大灵活性，spark支持在各种集群管理器（cluster manager）上运行，包括hadoop yarn、apache mesos，以及spark自带的一个简易调度器，叫做独立调度器。

1.5 spark特点

1、快：与hadoop的mapreduce相比，spark基于内存的运算要快上100倍以上，基于硬盘的运算也要快10倍以上。spark实现了高效的dag执行引擎，口头语通过基于内存来高效处理数据流。计算的中间结果是存在于内存中的。
2、易用：spark支持java、python和scala的api，还支持超过80种高级算法，使用户可以快速构建不同的应用。而且spark支持交互式的python和scala的shell，可以非常方便地在这些shell种使用spark集群来验证解决问题的方法。
3、通用：spark提供了统一的解决方案。spark可以用于，交互式查询（spark sql）、实时流处理（spark streaming）、机器学习（spark mllib）和图计算（graphx）。这些不同类型的处理1都可以在同一个应用种无缝使用。减少了开发和维护的人力成本和部署平台的物力成本。
4、兼容性：spark可以非常方便地与其它地开源产品进行融合。比如：spark可以使用hadoop的yarn和apache mesos作为它的资源管理和调度器，并且可以处理所有hadoop支持的数据，包括hdfs、hbase等。这对于已经部署hadoop集群的用户特别重要，因为不需要做任何数据迁移就可以使用spark的强大处理能力。

第 2 章：spark运行模式

部署spark集群大体上分为两种模式：单机模式与集群模式
大多数分布式框架都支持单机模式，方便开发者调试框架的运行环境。但是在生产环境种，并不会使用单机模式。因此，后续直接按照集群模式部署spark集群。
下面详细列举了spark目前支持的部署模式。
1、local模式：在本地部署spark服务
2、standalone模式：spark自带的任务调度模式。（国内常用）
3、yarn模式：spark使用hadoop的yarn组件进行资源和任务调度。（国内最常用）
4、mesos模式：spark使用mesos平台进行资源与任务的调度。（国内很少用）

2.2 local模式

local模式就是运行在一台计算机上的模式，通常就是用于在本机上练手和测试

2.2.1 安装使用

1）上传并解压spark安装包

[atguigu@hadoop102 sorfware]$ tar -zxvf spark-3.1.3-bin-hadoop3.2.tgz -C /opt/module/
[atguigu@hadoop102 module]$ mv spark-3.1.3-bin-hadoop3.2 spark-local

1
2
3

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

1
2
3
4
5
6

可以查看spark-submit所用参数

[atguigu@hadoop102 spark-local]$ bin/spark-submit
1

–class：表示要执行程序的主类
–master local[2]“
（1）local：没有指定线程数，则所有计算都运行在一个线程当中，没有任何并行计算。
（2）local[k]：指定使用k个core来运行计算，比如local[2]就是运行2个core来执行

20/09/20 09:30:53 INFO TaskSetManager:
20/09/15 10:15:00 INFO Executor: Running task 1.0 in stage 0.0 (TID 1)
20/09/15 10:15:00 INFO Executor: Running task 0.0 in stage 0.0 (TID 0)

1
2
3
4

（3）local[*]：默认模式。自动帮你按照cpu最多核来设置线程数。比如cpu有8核，spark帮你自动设置8个线程。

20/09/20 09:30:53 INFO TaskSetManager:
20/09/15 10:15:58 INFO Executor: Running task 1.0 in stage 0.0 (TID 1)
20/09/15 10:15:58 INFO Executor: Running task 0.0 in stage 0.0 (TID 0)
20/09/15 10:15:58 INFO Executor: Running task 2.0 in stage 0.0 (TID 2)
20/09/15 10:15:58 INFO Executor: Running task 4.0 in stage 0.0 (TID 4)
20/09/15 10:15:58 INFO Executor: Running task 3.0 in stage 0.0 (TID 3)
20/09/15 10:15:58 INFO Executor: Running task 5.0 in stage 0.0 (TID 5)
20/09/15 10:15:59 INFO Executor: Running task 7.0 in stage 0.0 (TID 7)
20/09/15 10:15:59 INFO Executor: Running task 6.0 in stage 0.0 (TID 6)

1
2
3
4
5
6
7
8
9
10

3）结果展示
该算法是利用蒙特-卡罗算法求pi

2.2.2 官方wordcount案例

1、需求：读取多个输入文件，统计每个单词出现的总次数。
2、需求分析

3、代码实现
1）准备文件

[atguigu@hadoop102 spark-local]$ mkdir input
1

在Input下创建2个文件1.txt和2.txt，并输入一下内容

hello atguigu
hello spark

1
2
3

2）启动spark-shell

[atguigu@hadoop102 spark-local]$ bin/spark-shell

20/07/02 10:17:11 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-1593656236294).
Spark session available as 'spark'.
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 3.1.3
      /_/
         
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>

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

注意：sc是sparkcore程序的入口；spark是sparksql程序入口；master=local[*]表示本地模式运行。
3）再开启一个hadoop102远程连接窗口，发现了一个sparksubmit进程

[atguigu@hadoop102 spark-local]$ jps
3627 SparkSubmit
4047 Jps

1
2
3
4

运行任务方式说明：spark-submit，是将jar上传到集群，执行spark任务；spark-shell，相当于命令行工具，本身也是一个application。
4）登录hadoop102:4040，查看程序运行情况

说明：本地模式下，默认的调度器为fifo。
5）运行workcount程序

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

res0: Array[(String, Int)] = Array((hello,4), (atguigu,2), (spark,2))

1
2
3
4

注意：只有collect开始执行时，才会加载数据
可登录hadoop102:4040查看程序运行结果

2.3 standalone模式

standalone模式是spark自带的资源调度引擎，构建一个由master+worker构成的spark集群，spark运行在集群种。
这个要和hadoop中的standalone区别开来。这里的standalone是指只用spark来搭建一个集群，不需要借助hadoop的yarn和mesos等其它框架。

2.3.1 master和worker集群资源管理

master：spark特有资源调度系统的leader。掌管着整个集群的资源信息，类似于yarn框架中的resourcemanager。
worker：spark特有资源调度系统的slave，有多个。每个slave掌管着所在节点的资源信息，类似于yarn框架中的nodemanager。

master和worker是spark的守护进程、集群资源管理者，即spark在特定模式（standalone）下正常运行必须要有的后台常驻进程。

2.3.2 driber和executor任务的管理者

driver和executor是临时程序，当有具体任务提交到spark集群才会开启的程序。standalone模式是spark自带的资源调度引擎，构建一个由master+worker构成spark集群，spark运行在集群中。
这个要和hadoop中的standalone区别开来。这里的standalone是指只用spark来搭建一个集群，不需要借助hadoop的yarn和mesos等其它框架。

2.3.2 安装使用

1、集群规划

2、再解压一份spark安装包，并修改解压后的文件夹名称为spark-standalone

[atguigu@hadoop102 sorfware]$ tar -zxvf spark-3.1.3-bin-hadoop3.2.tgz -C /opt/module/
[atguigu@hadoop102 module]$ mv spark-3.1.3-bin-hadoop3.2 spark-standalone

1
2
3

3、进入spark的配置文件/opt/module/spark-standalone/conf

[atguigu@hadoop102 spark-standalone]$ cd conf
1

4、修改slave文件，添加work节点

atguigu@hadoop102 conf]$ mv slaves.template slaves
[atguigu@hadoop102 conf]$ vim slaves
hadoop102
hadoop103
hadoop104

1
2
3
4
5
6

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

1
2
3
4
5
6

6、分发spark-standalone包

[atguigu@hadoop102 module]$ xsync spark-standalone/
1

7、启动spark集群

[atguigu@hadoop102 module]$ xsync spark-standalone/
1

查看三台服务器运行进程（xcall.sh是以前数仓项目里面讲的脚本）

[atguigu@hadoop102 spark-standalone]$ xcall.sh jps
================atguigu@hadoop102================
3238 Worker
3163 Master
================atguigu@hadoop103================
2908 Worker
================atguigu@hadoop104================
2978 Worker

1
2
3
4
5
6
7
8
9

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

export JAVA_HOME=XXXX
1

8、网页查看：hadoop102:8080
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.1.3.jar \
10

1
2
3
4
5
6

参数：–master spark://hadoop102:7077指定要连接的集群的master。
10、页面查看http://hadoop102:8080/，发现执行本次任务，默认采用三台服务器节点的总核数24核，每个节点内存1024M.
8080：master的webui
4040：application的webui的端口号

2.3.3 参数说明

1、配置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.1.3.jar \
10

1
2
3
4
5
6
7
8

2、页面查看http://hadoop102:8080/

3、基本语法

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

1
2
3
4
5
6
7

4、参数说明

2.3.4 配置历史服务

由于spark-shell停止掉后，hadoop102:4040页面就看不到历史任务的运行情况，所以开发时都配置历史服务器记录任务运行情况
1、修改spark-default.conf.template名称

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

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

[atguigu@hadoop102 conf]$ vim spark-defaults.conf
spark.eventLog.enabled          true
spark.eventLog.dir              hdfs://hadoop102:8020/directory

1
2
3
4

注意：需要启动hdaoop集群，hdfs上的目录需要提前存在

[atguigu@hadoop102 hadoop-3.1.3]$ sbin/start-dfs.sh
[atguigu@hadoop102 hadoop-3.1.3]$ hadoop fs -mkdir /directory

1
2
3

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

[atguigu@hadoop102 conf]$ vim 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
2
3
4
5
6
7

1）参数1含义：webui访问的端口号为18080
2）参数2含义：指定历史服务器日志存储路径（读）
3）参数3含义：指定保存application历史记录的个数，如果超过这个值，旧的应用程序信息将被删除，这个是内存中的应用数，而不是页面上的显示的应用数
4、分发配置文件

[atguigu@hadoop102 conf]$ xsync spark-defaults.conf spark-env.sh
1

5、启动历史服务

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

1
2
3

6、再次执行任务

[atguigu@hadoop102 spark-standalone]$ 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.12-3.1.3.jar \
10

1
2
3
4
5
6
7
8

7、查看spark历史服务地址：hadoop102:18080

2.3.5 配置高可用（HA）

1、高可用原理

2、配置高可用
1）停止集群

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

2）zookeeper正常安装并启动（基于以前讲的数仓项目脚本）

[atguigu@hadoop102 zookeeper-3.4.10]$ zk.sh start
1

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

[atguigu@hadoop102 conf]$ vim 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冲突
export SPARK_MASTER_WEBUI_PORT=8989

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

4）分发配置文件

[atguigu@hadoop102 conf]$ xsync spark-env.sh
1

5）在hadoop102上启动全部节点

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

6）在hadoop103上单独启动master节点

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

7）在启动一个hadoop102窗口，将/opt/module/spark-local/input数据上传到hadoop集群的/input目录

[atguigu@hadoop102 spark-standalone]$ hadoop fs -put /opt/module/spark-local/input/ /input
1

8）spark ha集群访问

[atguigu@hadoop102 spark-standalone]$
bin/spark-shell \
--master spark://hadoop102:7077,hadoop103:7077 \
--executor-memory 2g \
--total-executor-cores 2

1
2
3
4
5
6

参数：–master spark://hadoop102:7077指定要连接的集群的master
注：一旦配置了高可用以后，master后面要连接多个master
9）执行wordcount程序

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

res0: Array[(String, Int)] = Array((hello,4), (atguigu,2), (spark,2))

1
2
3
4

3、高可用性测试

1）查看hadoop102的master进程

[atguigu@hadoop102 ~]$ jps
5506 Worker
5394 Master
5731 SparkSubmit
4869 QuorumPeerMain
5991 Jps
5831 CoarseGrainedExecutorBackend

1
2
3
4
5
6
7
8

2）kill掉hadoop102的master进程，页面中观察http://hadoop103:8080/的状态是否切换为active

[atguigu@hadoop102 ~]$ kill -9 5394
1

3）再启动hadoop102的master进程

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

2.3.5 运行流程

spark由standalone-client核standalone-cluster两种模式，主要区别在于：driver程序的运行节点。
1、客户端模式

[atguigu@hadoop102 spark-standalone]$ 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.12-3.1.3.jar \
10

1
2
3
4
5
6
7
8
9

–deploy-mode client，表示driver程序运行再本地客户端，默认模式。
standalone client运行流程

2、集群模式

[atguigu@hadoop102 spark-standalone]$ 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.12-3.1.3.jar \
10

1
2
3
4
5
6
7
8
9

–deploy-mode cluster，表示driver程序运行在集群
standalone cluster运行流程

1）查看http://hadoop102:8989/页面，点击completed drivers里面的worker

2）跳转到spark worker页面，点击finished drivers中logs下面的stdout

3）最终打印结果如下

注意：在测试standalone模式，cluster运行流程的时候，阿里云用户访问不到worker，因为worker是从master内部跳转的，这是正常的，实际工作中我们不可能通过客户端访问的，这些恶端口都对外都会禁用，需要的时候会通过授权到master访问worker

2.4 yarn模式（重点）

spark客户端直接连接yarn，不需要额外构建spark集群

2.4.1 安装使用

1、停止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
2
3
4

2、为了防止和standalone模式冲突，再单独解压一份spark

[atguigu@hadoop102 software]$ tar -zxvf spark-3.1.3-bin-hadoop3.2.tgz -C /opt/module/
1

3、进入到/opt/module目录，修改spark-~名称为spark-yarn

[atguigu@hadoop102 module]$ mv spark-3.1.3-bin-hadoop3.2/ spark-yarn
1

4、修改hadoop配置文件/opt/module/~/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>

1
2
3
4
5
6
7
8
9
10
11
12
13

5、分发配置文件

[atguigu@hadoop102 conf]$ xsync /opt/module/hadoop-3.1.3/etc/hadoop/yarn-site.xml
1

6、修改/opt/~/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

1
2
3
4
5

7、启动hdfs以及yarn集群

[atguigu@hadoop102 hadoop-3.1.3]$ sbin/start-dfs.sh
[atguigu@hadoop103 hadoop-3.1.3]$ sbin/start-yarn.sh

1
2
3

8、执行一个程序

[atguigu@hadoop102 spark-yarn]$ bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master yarn \
./examples/jars/spark-examples_2.12-3.1.3.jar \
10

1
2
3
4
5
6

参数：–master yarn，表示yarn方式运行；–deploy-mode，表示客户端方式运行程序
9、查看hadoop103:8088页面，点击history，查看历史页面

2.4.2 配置历史服务

由于是重新解压的spark压缩文件，所以需要针对yarn模式，再次配置一下历史服务器。
1、修改spark-default.conf.template名称

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

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

参数1含义：webui访问的端口号为18080
参数2含义：指定历史服务器日志存储路径（读）
参数3含义：指定保存application历史记录的个数，如果超过这个值，旧的应用程序信息将被删除，这个是内存中的应用数，而不是页面上显示的应用数

2.4.3 配置查看历史日志

为了能从yarn上关联到spark历史服务器，需要配置spark历史服务器关联路径
目的：点击yarn（8088）上spark任务的history按钮，进入的是spark历史服务器（18080），而不再是yarn历史服务器（19888）
1、修改配置文件/opt/module/~/spark-defaults.conf
添加如下内容：

spark.yarn.historyserver.address=hadoop102:18080
spark.history.ui.port=18080
1
2

2、重启spark历史服务

[atguigu@hadoop102 spark-yarn]$ sbin/stop-history-server.sh 

[atguigu@hadoop102 spark-yarn]$ sbin/start-history-server.sh 

1
2
3
4

3、提交任务到yarn执行

[atguigu@hadoop102 spark-yarn]$ bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master yarn \
./examples/jars/spark-examples_2.12-3.1.3.jar \
10

1
2
3
4
5
6

4、web页面查看日志：http://hadoop103:8088/cluster

点击”history“跳转到http://hadoop102:18080/

2.4.4 运行流程

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 \
--master yarn \
--deploy-mode client \
./examples/jars/spark-examples_2.12-3.1.3.jar \
10

1
2
3
4
5
6
7

yarnclient运行模式介绍

2、集群模式

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

1
2
3
4
5
6
7

（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>

1
2
3
4
5

注意：hadoop历史服务器也要启动 mr-jobhistory-daemon.sh start historyserver
yarncluster模式

2.6 几种模式对比

2.7 端口号总结

1、spark查看当前spark-shell运行任务情况端口号：4040
2、spark master内部通信服务端口号：7077（类似于yarn的8032（rm和nm的内部通信）端口）
3、spark standalone模式master web端口号：8080（类似于hadoop yarn任务运行情况查看端口号：8088）（yarn模式）8989
4、spark历史服务器端口号：18080（类似于hadoop历史服务器端口号：19888）

第 3 章：workcount案例实操

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

3.1 部署环境

1、创建一个maven项目wordcount
2、在项目wordcount上点击右键，add framework support -> 勾选scala
3、在main下创建scala文件夹，并右键mark directory as sources root -> 在scala下创建包com.atguigu.spark
4、输入文件夹准备

5、导入项目依赖
下方的的是scala语言打包插件，只要使用scala语法打包运行到linux上面，必须要有

<dependencies>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-core_2.12</artifactId>
        <version>3.1.3</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>

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

3.2 本地调试

本地spark程序调试需要使用local提交模式，即将本机当作运行环境，master和worker都为本机。运行时直接加断点调试即可。如下：
1、代码实现

package com.atguigu.spark

import org.apache.spark.rdd.RDD
import org.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 atguigu
        val lineRdd: RDD[String] = sc.textFile("input")

        //4.读取的一行一行的数据分解成一个一个的单词（扁平化）(hello)(atguigu)(atguigu)
        val wordRdd: RDD[String] = lineRdd.flatMap(_.split(" "))

        //5. 将数据转换结构：(hello,1)(atguigu,1)(atguigu,1)
        val wordToOneRdd: RDD[(String, Int)] = wordRdd.map((_, 1))

        //6.将转换结构后的数据进行聚合处理 atguigu:1、1 =》1+1  (atguigu,2)
        val wordToSumRdd: RDD[(String, Int)] = wordToOneRdd.reduceByKey(_+_)

        //7.将统计结果采集到控制台打印
        wordToSumRdd.collect().foreach(println)

        //8.关闭连接
        sc.stop()
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

2、调试流程

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

3、集群运行

3.3 集群运行

1、修改代码，修改运行模式，将输出的方法修改为落盘，同时设置可以自定义的传入传出路径

package com.atguigu.spark

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

object WordCount {
  def main(args: Array[String]): Unit = {
    // 创建配置对象 添加配置参数
    val conf: SparkConf = new SparkConf()
      .setAppName("wc")
      // 如果是yarn模式  写yarn
      // 如果是本地模式一定要写local
    .setMaster("yarn")

    // 初始化sc
    val sc = new SparkContext(conf)

    // 编写wordCount计算流程
    // 把读入和写出的路径  做成动态的参数  可以由用户手动填写
    // 写成main方法参数
    val lineRDD: RDD[String] = sc.textFile(args(0))

    // 切分
    val wordRDD: RDD[String] = lineRDD.flatMap(_.split(" "))

    // 转换
    val tupleOneRDD: RDD[(String, Int)] = wordRDD.map((_, 1))

    // 聚合
    val wordCountRDD: RDD[(String, Int)] = tupleOneRDD.reduceByKey(_ + _)

    // 触发计算  一定要使用行动算子
    // 将结果保存到文件中
    // 不能重复写入同一个路径
    wordCountRDD.saveAsTextFile(args(1))
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

2、打包到集群测试
1）点击package打包，然后，查看打包完后的jar包

2）将wordcount.jar上传到/opt/module/spark-yarn目录
3）在hdfs上创建，存储输入文件的路径/input

[atguigu@hadoop102 spark-yarn]$ hadoop fs -mkdir /input
1

4）上传输入文件到/input路径

[atguigu@hadoop102 spark-yarn]$ hadoop fs -put /opt/module/spark-local/input/1.txt /input
1

5）执行任务

[atguigu@hadoop102 spark-yarn]$ bin/spark-submit \
--class com.atguigu.spark.WordCount \
--master yarn \
./WordCount.jar \
hdfs://hadoop102:8020/input \
hdfs://hadoop102:8020/output

1
2
3
4
5
6
7

注意：input和output都是hdfs上的集群路径
6）查看运行结果

[atguigu@hadoop102 spark-yarn]$ hadoop fs -cat /output/*
1

3.4 关联源码

1、按住ctrl键，点击rdd

2、提示下载或者绑定源码

3、解压资料包中spark-3.1.3.tgz到非中文路径。例如解压到：e:\02_software
4、点击attach source…按钮，选择源码路径e:\02_software\spark-3.1.3

3.5 异常处理

如果本机操作系统是windows，如果在程序中使用了hadoop相关的东西，比如写入文件到hdfs，则会遇到如下异常：

出现这个问题的原因，并不是程序的错误，而是用到了hadoop相关的服务，解决办法
1、配置hadoop_home环境变量
2、在idea中配置 run configuration，添加hadoop_home变量