大数据技术之Spark

1、Spark先容

1.1、Spark是什么

Spark是什么定义：Apache Spark是用于大规模数据（large-scala data）处理的同一（unified）分析引擎。

Spark最早源于一篇论文 Resilient Distributed Datasets: A Fault-Tolerant Abstraction for In-Memory Cluster Computing， 该论文是由加州大学柏克莱分校的 Matei Zaharia 等人发表的。论文中提出了一种弹性分布式数据集（即 RDD）的概念。

翻译过来就是：RDD 是一种分布式内存抽象，其使得步伐员可以大概在大规模集群中做内存运算，而且有一定的容错方式。而这也是整个Spark的核心数据结构，Spark 整个平台都围绕着RDD进行。

简而言之，Spark 借鉴了 MapReduce 思想发展而来，保留了其分布式并行计算的优点并改进了其明显的缺陷。让中心数据存储在内存中提 高了运行速率、并提供丰富的操作数据的API进步了开发速率。

• Spark是一款分布式内存计算的同一分析引擎。 其特点就是对恣意类型的数据进行自定义计算。
  
• Spark可以计算：结构化、半结构化、非结构化等各种类型的数据结构，同时也支持使用Python、Java、Scala、R以及SQL语言去开发应用 步伐计算数据。
  
• Spark的适用面非常广泛，以是，被称之为 同一的（适用面广）的分析引擎（数据处理）
  

1.2、Spark风雨十年

Spark 是加州大学伯克利分校AMP实行室（Algorithms Machines and People Lab）开发的通用大数据处理框架。 Spark的发展历史，履历过几大重要阶段，如下图所示：

Stack Overflow的数据可以看出，2015年开始Spark每月的问题提交数目已经逾越Hadoop，而2018年Spark Python版本的APIPySpark每月的问题提交数目也已超过Hadoop。2019年排名Spark第一，PySpark第二；而十年的累计排名是Spark第一，PySpark第 三。按照这个趋势发展下去，Spark和PySpark在未来很长一段时间内应该还会处于垄断地位。

十年走来，Spark目前已经迭代到了3.2.0版本（2021.10.13发布），本次课程基于最新的Spark 3.2.0版本进行授课

1.3、Spark VS Hadoop(MapReduce)

Spark和前面学习的Hadoop技术栈有何区别呢？

尽管Spark相对于Hadoop而言具有较大优势，但Spark并不能完全替代Hadoop

• 在计算层面，Spark相比较MR（MapReduce）有巨大的性能优势，但至今仍有很多计算工具基于MR构架，比如非常成熟的Hive
  
• Spark仅做计算，而Hadoop生态圈不仅有计算（MR）也有存储（HDFS）和资源管理调理（YARN），HDFS和YARN仍是很多大数据 体系的核心架构。
  

1.4、Spark四大特点

速率快：由于Apache Spark支持内存计算，而且通过DAG（有向无环图）实行引擎支持无环数据流，以是官方宣称其在内存中的运算速率要比Hadoop的MapReduce快100倍，在硬盘中要快10倍。

Spark处理数据与MapReduce处理数据相比，有如下两个不同点：

• 其一、Spark处理数据时，可以将中心处理结果数据存储到内存中；
  
• 其二、Spark 提供了非常丰富的算子(API), 可以做到复杂使命在一个Spark 步伐中完成.
  
  

易于使用：Spark 的版本已经更新到 Spark 3.2.0（截止日期2021.10.13），支持了包括 Java、Scala、Python 、R和SQL语言在内的多种语言。为了 兼容Spark2.x企业级应用场景，Spark仍然持续更新Spark2版本。

通用性强：在 Spark 的基础上，Spark 还提供了包括Spark SQL、Spark Streaming、MLib 及GraphX在内的多个工具库，我们可以在一个应用中无缝 地使用这些工具库。

运行方式:Spark 支持多种运行方式，包括在 Hadoop 和 Mesos 上，也支持 Standalone的独立运行模式，同时也可以运行在云Kubernetes（Spark2.3开始支持）上。

对于数据源而言，Spark 支持从HDFS、HBase、Cassandra 及 Kafka 等多种途径获取数据。
1.5、Spark 框架模块-了解

整个Spark 框架模块包罗：Spark Core、 Spark SQL、 Spark Streaming、 Spark GraphX、 Spark MLlib，而后四项的本领都是建立在核心引擎之上

Spark Core：Spark的核心，Spark核心功能均由Spark Core模块提供，是Spark运行的基础。Spark Core以RDD为数据抽象，提供Python、Java、 Scala、R语言的API，可以编程进行海量离线数据批处理计算。
SparkSQL：基于SparkCore之上，提供结构化数据的处理模块。SparkSQL支持以SQL语言对数据进行处理，SparkSQL自己针对离线计算场景。同 时基于SparkSQL，Spark提供了StructuredStreaming模块，可以以SparkSQL为基础，进行数据的流式计算。
SparkStreaming：以SparkCore为基础，提供数据的流式计算功能。
MLlib：以SparkCore为基础，进行机器学习计算，内置了大量的机器学习库和API算法等。方便用户以分布式计算的模式进行机器学习计算。
GraphX：以SparkCore为基础，进行图计算，提供了大量的图计算API，方便用于以分布式计算模式进行图计算。
1.6、Spark的运行模式 - 了解

Spark提供多种运行模式，包括：

• 本地模式（单机）本地模式就是以一个独立的进程，通过其内部的多个线程来模拟整个Spark运行时环境
  

• Standalone模式（集群）Spark中的各个角色以独立进程的情势存在，并构成Spark集群环境
  

• Hadoop YARN模式（集群）Spark中的各个角色运行在YARN的容器内部，并构成Spark集群环境
  
• Kubernetes模式（容器集群）Spark中的各个角色运行在Kubernetes的容器内部，并构成Spark集群环境
  
• 云服务模式（运行在云平台上）......
  

1.7、Spark的架构角色 - 理解

YARN重要有4类角色，从2个层面去看

Spark运行角色

注意：正常环境下Executor是干活的角色，不过在Local模式下，Driver即管理又干活
2、Spark环境搭建-Local

2.1、课程服务器环境

1. 本次课程使用三台Linux虚拟机服务器来学习, 三台虚拟机的功能分配是:  
2. hadoop102: Master和 Worker
3. hadoop103: Worker
4. hadoop104: Worker

复制代码

2.2、根本原理

本质：启动一个JVM Process进程(一个进程内里有多个线程)，实行使命Task
Local模式可以限制模拟Spark集群环境的线程数目, 即Local[N] 或 Local

此中N代表可以使用N个线程，每个线程拥有一个cpu core。如果不指定N， 则默认是1个线程（该线程有1个core）。 通常Cpu有几个Core，就指定几个 线程，最大化利用计算本领.*
如果是local

，则代表 Run Spark locally with as many worker threads as logical cores on your machine.按照Cpu最多的Cores设置线程数
Local 下的角色分布：
资源管理：

• Master：Local进程自己
  
• Worker：Local进程自己
  

使命实行：

• Driver：Local进程自己
  
• Executor：不存在，没有独立的Executor角色, 由Local进程(也就是Driver)内的线程提供计算本领
  

PS: Driver也算一种特别的Executor, 只不过多数时间, 我们将Executor当做纯Worker对待, 这样和Driver好区分(一类是管理 一类是工人)
2.3、搭建环境

• 开箱即用：直接启动bin目次下的
  
  1. spark-shell:/opt/module/spark/bin/spark-shell

  复制代码
  运行成功以后，有如下提示信息：
  

• sc：SparkContext实例对象：
  
• spark：SparkSession实例对象
  
• 4040：Web监控页面端标语
  

2.4、基于bin/pyspark

1. bin/pyspark

复制代码

步伐, 可以提供一个 交互式的 Python解释器环境, 在这内里可以用Python语言调用Spark API 进行计算

1. 示例代码 ,将数组内容都+进行 计 算 :  
2. sc.parallelize([1,2,3,4,5]).map(lambda x: x + 1).collect()

复制代码

4040端口是一个WEBUI端口, 可以在浏览器内打开, 输入:服务器ip:4040 即可打开

打开监控页面后, 可以发现 在步伐内仅有一个Driver由于我们是Local模式, Driver即管理又干活. 同时, 输入jps可以看到local模式下的唯一进程存在 这个进程 即是master也是worker
2.5、基于bin/spark-submit测试

bin/spark-submit步伐, 作用: 提交指定的Spark代码到Spark环境中运行

1. ./spark-submit ../examples/src/main/python/pi.py  10

复制代码

pyspark/spark-shell/spark-submit 对比

2.6、基于spark-shell

1. sc.parallelize(Array(1,2,3,4,5)).map(x=> x + 1).collect()

复制代码

3、Spark环境搭建-Standalone

3.1、Standalone 架构

Standalone模式是Spark自带的一种集群模式，不同于前面本地模式启动多个进程来模拟集群的环境，Standalone模 式是真实地在多个机器之间搭建Spark集群的环境，完全可以利用该模式搭建多机器集群，用于实际的大数据处理。
StandAlone 是完整的Spark运行环境,此中:
Master角色以Master进程存在, Worker角色以Worker进程存在
Driver角色运行时间有大概存在Master节点上，也有大概不存在一起，Executor运行于Worker进程内,属于Worker一个子进程 由Worker提供资源供给它们运行

StandAlone集群在进程上重要有3类进程:
主节点Master进程：

• Master角色, 管理整个集群资源，并托管运行各个使命的Driver
  

从节点Workers：

• Worker角色, 管理每个机器的资源，分配对应的资源来运行Executor(Task)；每个从节点分配资源信息给Worker管理，资源信息包罗内存Memory和CPU Cores核数
  

历史服务器HistoryServer(可选)：
Spark Application运行完成以后，生存变乱日记数据至HDFS，启动HistoryServer可以查看应用运行相干信息。

3.2、环境搭建

3.2.1、安装Spark

上传Spark到/opt/software下

1. tar -zxvf spark-3.2.0-bin-hadoop3.2.tgz -C /opt/module

复制代码

解压到/opt/module下，

进入到/opt/module下，把文件夹重名为spark

1. mv spark-3.2.0-bin-hadoop3.2/ spark

复制代码

3.2.2、安装JDK和Python

JDK前提安装完毕
上传Anaconda3-2021.05-Linux-x86_64.sh

安装Anaconda3-2021.05-Linux-x86_64.sh

1. bash Anaconda3-2021.05-Linux-x86_64.sh

复制代码

之间回车就可以

输入Yes

输入安装路径

输入yes进行初始化
退出机器，重新进入

3.2.3、设置国内源

如果你安装好后, 可以打开:vim ~/.condarc这个文件, 追加如下内容:

1. channels:
2.   - defaults
3. show_channel_urls: true
4. default_channels:
5.   - https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/main
6.   - https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/r
7.   - https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/msys2
8. custom_channels:
9.   conda-forge: https://mirrors.tuna.tsinghua.edu.cn/anaconda/cloud
10.   msys2: https://mirrors.tuna.tsinghua.edu.cn/anaconda/cloud
11.   bioconda: https://mirrors.tuna.tsinghua.edu.cn/anaconda/cloud
12.   menpo: https://mirrors.tuna.tsinghua.edu.cn/anaconda/cloud
13.   pytorch: https://mirrors.tuna.tsinghua.edu.cn/anaconda/cloud
14.   simpleitk: https://mirrors.tuna.tsinghua.edu.cn/anaconda/cloud

复制代码

3.2.4、创建捏造环境

1. conda env list
2. conda create -n pyspark python=3.8 #创建python3.8.8环境
3. conda activate pyspark   #激活环境
4. conda deactivate pyspark #退出环境

复制代码

3.3、环境变量

1. -  SPARK_HOME: 表示Spark安装路径在哪里
2. -  PYSPARK_PYTHON: 表示Spark想运行Python程序, 那么去哪里找python执行器
3. -  JAVA_HOME: 告知Spark Java在哪里
4. -  HADOOP_CONF_DIR: 告知Spark Hadoop的配置文件在哪里
5. -  HADOOP_HOME: 告知Spark Hadoop安装在哪里

复制代码

3.3.1、设置Spark

编辑 vim /etc/profile.d/my_env.sh

1. #SPARK_HOME
2. export SPARK_HOME=/opt/module/spark
3. export PYSPARK_PYTHON=/opt/module/anaconda3/envs/pyspark/bin/python3.8
4. export HADOOP_CONF_DIR=$HADOOP_HOME/etc/hadoop

复制代码

编辑 vim ~/.bashrc

1. export JAVA_HOME=/opt/module/jdk1.8.0_212
2. export PYSPARK_PYTHON=/opt/module/anaconda3/envs/pyspark/bin/python3.8

复制代码

3.4、StandAlone部署

3.4.1、集群规划

1. hadoop102运行: Spark的Master进程  和 1个Worker进程
2. hadoop103运行: spark的1个worker进程
3. hadoop104运行: spark的1个worker进程
4. 整个集群提供: 1个master进程 和 3个worker进程

复制代码

3.4.2、在所有机器安装Anaconda3

3.4.3、编辑设置文件

进入到/opt/module/spark

• workers
  

1. # 改名
2. mv workers.template workers
3. # 编辑worker文件
4. vim workers
5. 将里面的localhost删除, 追加
6. hadoop102
7. hadoop103
8. hadoop104

复制代码

• 设置spark-env.sh文件
  

1. # 1. 改名
2. mv spark-env.sh.template spark-env.sh

复制代码

1. # 2. 编辑spark-env.sh, 在底部追加如下内容
3. ## 设置JAVA安装目录
4. JAVA_HOME=/opt/module/jdk1.8.0_212
6. ## HADOOP软件配置文件目录，读取HDFS上文件和运行YARN集群
7. HADOOP_CONF_DIR=/opt/module/hadoop-3.1.3/etc/hadoop
8. YARN_CONF_DIR=/opt/module/hadoop-3.1.3/etc/hadoop
10. ## 指定spark老大Master的IP和提交任务的通信端口
11. # 告知Spark的master运行在哪个机器上
12. export SPARK_MASTER_HOST=hadoop102
13. # 告知sparkmaster的通讯端口
14. export SPARK_MASTER_PORT=7077
15. # 告知spark master的 webui端口
16. SPARK_MASTER_WEBUI_PORT=8080
18. # worker cpu可用核数
19. SPARK_WORKER_CORES=1
20. # worker可用内存
21. SPARK_WORKER_MEMORY=1g
22. # worker的工作通讯地址
23. SPARK_WORKER_PORT=7078
24. # worker的 webui地址
25. SPARK_WORKER_WEBUI_PORT=8081
27. ## 设置历史服务器
28. # 配置的意思是  将spark程序运行的历史日志 存到hdfs的/sparklog文件夹中
29. SPARK_HISTORY_OPTS="-Dspark.history.fs.logDirectory=hdfs://hadoop102:9820/sparklog/ -Dspark.history.fs.cleaner.enabled=true"

复制代码

在HDFS上创建步伐运行历史记录存放的文件夹:

1. hadoop fs -mkdir /sparklog
2. hadoop fs -chmod 777 /sparklog

复制代码

• 设置spark-defaults.conf文件
  

1. # 1. 改名
2. mv spark-defaults.conf.template spark-defaults.conf

复制代码

1. # 2. 修改内容, 追加如下内容
2. # 开启spark的日期记录功能
3. spark.eventLog.enabled         true
4. # 设置spark日志记录的路径
5. spark.eventLog.dir         hdfs://hadoop102:9820/sparklog/
6. # 设置spark日志是否启动压缩
7. spark.eventLog.compress         true

复制代码

• 设置log4j.properties（可选设置）
  

1. 1、mv log4j.properties.template log4j.properties

复制代码

1. 2、把INFO级别改成WARN级别

复制代码

3.4.4、启动历史服务器

进入到spark的sbin目次下

1. ./start-history-server.sh
3. 端口：18080

复制代码

3.4.5、启动集群

1. ./start-all.sh

复制代码

查看Master的WEB UI
默认端口master我们设置到了8080
如果端口被占用, 会顺延到8081 ...;8082... 8083... 直到申请到端口为止
可以在日记中查看, 详细顺延到哪个端口上:

3.4.5、集群测试

• bin/pyspark
  

1. bin/pyspark --master spark://hadoop102:7077
2. sc.parallelize([1,2,3,4,5]).map(lambda x:x+1).collect()

复制代码

• bin/spark-shell
  

1. bin/spark-shell --master spark://hadoop102:7077sc.parallelize(Array(1,2,3,4,5)).map(x=> x + 1).collect()

复制代码

• bin/spark-submit (PI)
  

1. bin/spark-submit --master spark://hadoop102:7077 /opt/module/spark/examples/src/main/python/pi.py 100

复制代码

3.4.6、Spark应用架构

Spark 应用架构从图中可以看到Spark Application运行到集群上时，由两部门构成：Driver Program和Executors。
第一、Driver Program

• 相当于AppMaster，整个应用管理者，负责应用中所有Job的调理实行;
  
• 运行JVM Process，运行步伐的MAIN函数，必须创建SparkContext上下文对象；
  
• 一个SparkApplication仅有一个；
  

第二、Executors

• 相当于一个线程池，运行JVM Process，此中有很多线程，每个线程运行一个Task使命，一个Task使命运行需要1 Core CPU，所 有可以认为Executor中线程数就等于CPU Core核数；
  
• 一个Spark Application可以有多个，可以设置个数和资源信息；
  

1. 用户程序从最开始的提交到最终的计算执行，需要经历以下几个阶段：
3. 用户程序创建 SparkContext 时，新创建的 SparkContext 实例会连接到 ClusterManager。 Cluster Manager 会根据用户 提交时设置的 CPU 和内存等信息为本次提交分配计算资源，启动 Executor。
5. Driver会将用户程序划分为不同的执行阶段Stage，每个执行阶段Stage由一组完全相同Task组成，这些Task分别作用于待处 理数据的不同分区。在阶段划分完成和Task创建后， Driver会向Executor发送 Task；
7. Executor在接收到Task后，会下载Task的运行时依赖，在准备好Task的执行环境后，会开始执行Task，并且将Task的运行状态 汇报给Driver；
9. Driver会根据收到的Task的运行状态来处理不同的状态更新。 Task分为两种：一种是Shuffle Map Task，它实现数据的重新 洗牌，洗牌的结果保存到Executor 所在节点的文件系统中；另外一种是Result Task，它负责生成结果数据；
11. Driver 会不断地调用Task，将Task发送到Executor执行，在所有的Task 都正确执行或者超过执行次数的限制仍然没有执行成 功时停止；

复制代码

3.4.7、Spark步伐运行条理结构

在前面我们接触到了不少的监控页面,有4040,有8080,有18080,它们有何区别吗?

• 4040: 是一个运行的Application在运行的过程中临时绑定的端口,用以查看当前使命的状态.4040被占用会顺延到4041.4042等4040是一个临时端口,当前步伐运行完成后, 4040就会被注销
  
• 8080: 默认是StandAlone下, Master角色(进程)的WEB端口,用以查看当前Master(集群)的状态
  
• 18080: 默认是历史服务器的端口, 由于每个步伐运行完成后,4040端口就被注销了. 在以后想回看某个步伐的运行状态就可以通过历史 服务器查看,历史服务器恒久稳定运行,可供随时查看被记录的步伐的运行过程.
  

1. sc.textFile("hdfs://hadoop102:9820/word.txt").flatMap(lambda x:x.split(" ")).map(lambda x,(x,1)).reduceByKey(lambda a,b:a+b).collect()

复制代码

Spark Application步伐运行时三个核心概念：Job、Stage、 Task，分析如下：

• Job：由多个 Task 的并行计算部门，一样平常 Spark 中的action 操作（如 save、collect，背面进一步分析），会 生成一个 Job。
  
• Stage：Job的构成单元，一个Job会切分成多个 Stage，Stage 彼此之间相互依赖顺序实行，而每个 Stage是多个Task 的集合，类似map 和reduce stage。
  
• Task：被分配到各个 Executor 的单元工作内容，它是Spark 中的最小实行单元，一样平常来说有多少个 Paritition（物理层面的概念，即分支可以理解为将数据分别成不同 部门并行处理），就会有多少个 Task，每个 Task 只会处 理单一分支上的数据。
  

3.4.8、总结

StandAlone的原理?

1. Master和Worker角色以独立进程的形式存在,并组成Spark运行时环境(集群)

复制代码

Standalone如何提交Spark应用？

1. bin/spark-submit --master spark://server:7077

复制代码

4040\8080\18080分别是什么?

1. 4040是单个程序运行的时候绑定的端口可供查看本任务运行情况

复制代码

Job\State\Task的关系?

1. 一个Spark程序会被分成多个子任务(Job)运行, 每一个Job会分成多个State(阶段)来 运行, 每一个State内会分出来多个Task(线程)来执行具体任务

复制代码

4、Standalone HA

4.1、高可用HA

Spark Standalone集群是Master-Slaves架构的集群模式，和大部门的Master-Slaves结构集群一样，存在着Master单点故障（SPOF）的问题

如何解决这个单点故障的问题，Spark提供了两种方案：

• 基于文件体系的单点恢复(Single-Node Recovery with Local File System)--只能用于开发或测试环境。
  
• 基于zookeeper的Standby Masters(Standby Masters with ZooKeeper)--可以用于生产环境。ZooKeeper提供了一个Leader Election机制，利用这个机制可以包管虽然集群存在多个Master，但是只有一个是Active的，其他的都是Standby。当Active的Master出现故障时，另外的一个Standby Master会被推选出来。由于集群的信息，包括Worker， Driver和Application的信息都已经长期化到文件体系，因此在切换的过程中只会影响新Job的提交，对 于正在进行的Job没有任何的影响。参加ZooKeeper的集群整体架构如下图所示。
  
  
  
  
  

4.2、基于Zookeeper实现HA

在spark-env.sh中,删除

1. #export SPARK_MASTER_HOST=hadoop102

复制代码

在spark-env.sh中, 增长:

1. SPARK_DAEMON_JAVA_OPTS="-Dspark.deploy.recoveryMode=ZOOKEEPER -Dspark.deploy.zookeeper.url=hadoop102:2181,hadoop103:2181,hadoop104:2181 -Dspark.deploy.zookeeper.dir=/spark-ha"

复制代码

将spark-env.sh 分发到每一台服务器上

1. xsync spark-env.sh

复制代码

停止当前StandAlone集群

1. sbin/stop-all.sh

复制代码

启动集群:

1. # 在hadoop102上 启动一个master 和全部worker
2. sbin/start-all.sh
4. # 注意, 下面命令在hadoop103上执行
5. sbin/start-master.sh
6. # 在hadoop103上启动一个备用的master进程

复制代码

4.3、查看WebUI

1. http://hadoop102:8080/
2. http://hadoop103:8080/

复制代码

1. 如果将hadoop102的Master进程Kill掉，hadoop103的Master在1Min-2Min左右会接替hadoop102的Master作用。 也就是在执行过程中，使用jps  查看Active Master进程ID，将其kill，观察Master是否自动切换与应用运行完成结束。（需要等待1-2min）

复制代码

5、Spark On YARN 环境搭建

5.1、引言

按照前面环境部署中所学习的, 如果我们想要一个稳定的生产Spark环境, 那么最优的选择就是构建:HA StandAlone集群.
不过在企业中, 服务器的资源总是告急的, 很多企业不管做什么业务,都根本上会有Hadoop集群. 也就是会有YARN集群.
对于企业来说,在已有YARN集群的前提下在单独准备Spark StandAlone集群,对资源的利用就不高. 以是, 在企业中,多 数场景下,会将Spark运行到YARN集群中.
YARN自己是一个资源调理框架, 负责对运行在内部的计算框架进行资源调理管理. 作为典型的计算框架, Spark自己也是直接运行在YARN中, 并接受YARN的调理的.
以是, 对于Spark On YARN, 无需部署Spark集群, 只要找一台服务器, 充当Spark的客户端, 即可提交使命到YARN集群中运行.
5.2、本质

Spark On Yarn的本质?

• Master角色由YARN的ResourceManager担任
  
• Worker角色由YARN的NodeManager担任
  
• Driver角色运行在YARN容器内 或 提交使命的客户端进程
  
• 真正干活的Executor运行在YARN提供的容器内
  

5.3、设置

确保:

• HADOOP_CONF_DIR
  
• YARN_CONF_DIR
  

在spark-env.sh 以及 环境变量设置文件中即可
5.4、测试

Spark On YARN是有两种运行模式的,
一种是Cluster模式一种是Client模式.这两种模式的区别就是Driver运行的位置.
Cluster模式即river运行在YARN容器内部, 和ApplicationMaster在同一个容器内
Client模式即river运行在客户端进程中, 比如Driver运行在spark-submit步伐的进程中

如图, 此为Cluster模式Driver运行在容器内部

如图, 此为Client模式Driver运行在客户端步伐进程中(以spark-submit为例)

两种模式的区别

bin/pyspark

1. bin/pyspark --master yarn --deploy-mode client|cluster
2. # --deploy-mode 选项是指定部署模式, 默认是 客户端模式
3. # client就是客户端模式
4. # cluster就是集群模式
5. # --deploy-mode 仅可以用在YARN模式下

复制代码

  注意: 交互式环境 pyspark  和 spark-shell  无法运行 cluster模式
  bin/spark-shell

1. bin/spark-shell --master yarn --deploy-mode client|cluster

复制代码

  注意: 交互式环境 pyspark  和 spark-shell  无法运行 cluster模式
  bin/spark-submit (PI)

1. bin/spark-submit --master yarn --deploy-mode client|cluster /xxx/xxx/xxx.py 参数

复制代码

5.5、两种模式运行代码

1. 假设运行圆周率PI程序，采用client模式，命令如下：
2. bin/spark-submit
3. --master yarn
4. --deploy-mode client
5. --driver-memory 512m
6. --executor-memory 512m
7. --num-executors 1
8. --texecutor-cores 2
9. /examples/src/main/python/pi.py  10

复制代码

1. bin/spark-submit
2. --master yarn
3. --deploy-mode cluster
4. --driver-memory 512m
5. --executor-memory 512m
6. --num-executors 1
7. --executor-cores 2
8. /examples/src/main/python/pi.py  10

复制代码

5.6、Spark On Yarn两种模式总结

Client模式和Cluster模式最最本质的区别是：Driver步伐运行在哪里。

• Client模式：学习测试时使用，生产不保举(要用也可以,性能略低,稳定性略低)
  
  
  
  • Driver运行在Client上,和集群的通信成本高
    
  • Driver输出结果会在客户端显示
    
  
  
• Cluster模式：生产环境中使用该模式
  
  
  
  • Driver步伐在YARN集群中，和集群的通信成本低
    
  • Driver输出结果不能在客户端显示
    
  • 该模式下Driver运行ApplicattionMaster这个节点上,由Yarn管理，如果出现问题，yarn会重启 ApplicattionMaster(Driver)
    
  
  

在YARN Client模式下，Driver在使命提交的本地机器上运行，示意图如下

1. 1、Driver在任务提交的本地机器上运行，Driver启动后会和ResourceManager通讯申请启动ApplicationMaster
2. 2、随后ResourceManager分配Container，在合适的NodeManager上启动ApplicationMaster，此时的ApplicationMaster的功能相当于一个ExecutorLaucher，只负责向ResourceManager申请Executor内存；
3. 3、ResourceManager接到ApplicationMaster的资源申请后会分配Container，然后ApplicationMaster在资源分 配指定的NodeManager上启动Executor进程；
4. 4、Executor进程启动后会向Driver反向注册，Executor全部注册完成后Driver开始执行main函数；
5. 5、之后执行到Action算子时，触发一个Job，并根据宽依赖开始划分Stage，每个Stage生成对应的TaskSet，之后 将Task分发到各个Executor上执行。

复制代码

在YARN Cluster模式下，Driver运行在NodeManager Contanier中，此时Driver与AppMaster合为一体，示意图如

1. 1、任务提交后会和ResourceManager通讯申请启动ApplicationMaster;
2. 2、随后ResourceManager分配Container，在合适的NodeManager上启动ApplicationMaster，此时的ApplicationMaster就是Driver；
3. 3、Driver启动后向ResourceManager申请Executor内存，ResourceManager接到ApplicationMaster的资源申请 后会分配Container,然后在合适的NodeManager上启动Executor进程;
4. 4、Executor进程启动后会向Driver反向注册;
5. 5、Executor全部注册完成后Driver开始执行main函数，之后执行到Action算子时，触发一个job，并根据宽依赖开 始划分stage，每个stage生成对应的taskSet，之后将task分发到各个Executor上执行;

复制代码

6、Pyspark

我们前面使用过bin/pyspark 步伐, 要注意, 这个只是一个应用步伐, 提供一个Python解释器实行环境来运行Spark使命 我们现在说的PySpark, 指的是Python的运行类库, 是可以在Python代码中:import pyspark
PySpark 是Spark官方提供的一个Python类库, 内置了完全的Spark API, 可以通过PySpark类库来编写Spark应用步伐, 并将其提交到Spark集群中运行.下图是,PySpark类库和标准Spark框架的简单对比

6.1、在集群安装Pyspark

1. conda activate pyspark   #激活环境
3. pip install pyspark==3.2.0

复制代码

6.2、在本地安装Pyspark

1. pip install pyspark==3.2.0 -i https://pypi.douban.com/simple

复制代码

6.3、连接集群环境

通过pycharm--->file--->settings

1. /opt/module/anaconda3/envs/pyspark/bin/python

复制代码

7、PySpark开发入口

Spark Application步伐入口为：SparkContext，任何一个应用首先需要构建SparkContext对象，如下两步构建：

• 第一步、创建SparkConf对象设置Spark Application根本信息，比如应用的名称AppName和应用运行Master
  
• 第二步、基于SparkConf对象，创建SparkContext对象
  

文档：http://spark.apache.org/docs/3.1.2/rdd-programming-guide.html

7.1、WordCount

1. from pyspark import SparkConf, SparkContext
2. if __name__ == '__main__':
3.     conf = SparkConf().setMaster("yarn").set("spark.hadoop.fs.defaultFS", "hdfs://hadoop102:9820").setAppName("wc")
4.     sc = SparkContext(conf=conf)
5.     rdd = sc.parallelize([1, 2, 3, 4]).map(lambda x: x + 1).collect()
6.     print(rdd)

复制代码

1. # coding:utf8
2. from pyspark import SparkConf, SparkContext
5. if __name__ == '__main__':
6.     conf = SparkConf().setAppName("WordCountHelloWorld").setMaster("local[*]")
7.     # 通过SparkConf对象构建SparkContext对象
8.     sc = SparkContext(conf=conf)
10.     # 需求 : wordcount单词计数, 读取HDFS上的words.txt文件, 对其内部的单词统计出现 的数量
11.     # 读取文件
12.     file_rdd = sc.textFile("hdfs://hadoop102:9820/word.txt")
14.     # 将单词进行切割, 得到一个存储全部单词的集合对象
15.     words_rdd = file_rdd.flatMap(lambda line: line.split(" "))
17.     # 将单词转换为元组对象, key是单词, value是数字1
18.     words_with_one_rdd = words_rdd.map(lambda x: (x, 1))
20.     # 将元组的value 按照key来分组, 对所有的value执行聚合操作(相加)
21.     result_rdd = words_with_one_rdd.reduceByKey(lambda a, b: a + b)
23.     # 通过collect方法收集RDD的数据打印输出结果
24.     print(result_rdd.collect())

复制代码

原理分析

7.2、提交到集群运行

注意：去掉setMaster("local

")

1. from pyspark import SparkConf, SparkContext
4. if __name__ == '__main__':
5.     conf = SparkConf().setAppName("WordCountHelloWorld")
6.     # 通过SparkConf对象构建SparkContext对象
7.     sc = SparkContext(conf=conf)
9.     # 需求 : wordcount单词计数, 读取HDFS上的words.txt文件, 对其内部的单词统计出现 的数量
10.     # 读取文件
11.     file_rdd = sc.textFile("hdfs://hadoop102:9820/word.txt")
13.     # 将单词进行切割, 得到一个存储全部单词的集合对象
14.     words_rdd = file_rdd.flatMap(lambda line: line.split(" "))
16.     # 将单词转换为元组对象, key是单词, value是数字1
17.     words_with_one_rdd = words_rdd.map(lambda x: (x, 1))
19.     # 将元组的value 按照key来分组, 对所有的value执行聚合操作(相加)
20.     result_rdd = words_with_one_rdd.reduceByKey(lambda a, b: a + b)
22.     # 通过collect方法收集RDD的数据打印输出结果
23.     print(result_rdd.collect())

复制代码

1. bin/spark-submit --master yarn /tmp/pycharm_project_931/00_example/hw.py

复制代码

7.3、Python On Spark 实行原理

PySpark宗旨是在不粉碎Spark已有的运行时架构，在Spark架构外层包装一层Python API，借助Py4j实现Python和Java的交互，进而实现通过Python编写Spark应用步伐，其运行时架构如下图所示。

8、分布式代码实行分析

8.1、Spark 集群角色回顾(以YARN为例)

8.2、分布式代码实行分析

Spark Application应用步伐运行时，无论client还是cluster部署模式DeployMode，当Driver Program和Executors启动完成以后，就要开始实行应用步伐中MAIN函数的代码，以词频统计WordCount步伐为例剖析解说

第一、构建SparkContex对象和关闭SparkContext资源，都是在Driver Program中实行，上图中①和③都是，如 下图所示：

第二、上图中②的加载数据【A】、处理数据【B】和输出数据【C】代码，都在Executors上实行，从WEB UI监控 页面可以看到此Job（RDD#action触发一个Job）对应DAG图，如下所示：


免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。