Spark介绍与安装

19.Spark Core

一、Spark的介绍

1.定义

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

2.Spark VS MapReduce

Spark的诞生仅仅是为了替换早期的Hadoop的MapReduce计算引擎。Spark并没有存储解决方案，在Spark的架构中，底层存储方案依然延续Hadooop的HDFS/Hive/HBase.由于Hadoop的MapReduce是大数据时代出现的第一代大数据分析工具，因为出现比较早仅仅为了满足大数据计算的刚性需求（能够做到对大数据的计算，并且可以保证在一个合理的时间范围内）。因此伴随着科技的进步，互联网的快速发展，人们开始对大数据计算提出了更苛刻要求

Spark的设计中汲取了Map Reduce的设计经验，在2009年Spark在加州伯克利AMP实验室诞生，2010年首次开源，2013年6开始在Apache孵化，2014年2月份正式成为Apache顶级项目。由于Spark计算底层引擎使用批处理计算模型实现，非常容易被广大深受MapReduce计算折磨程序员所接受，所以就导致了Spark的关注度直线提升

Spark 计算比 MapReduce 快的根本原因在于 DAG（有向无环图） 计算模型。一般而言，DAG 相比MapReduce 在大多数情况下可以减少shuffle次数。Spark 的 DAGScheduler 相当于一个改进版的 MapReduce，如果计算不涉及与其他节点进行数据交换，Spark可以在内存中一次性完成这些操作，也就是中间结果无须落盘，减少了磁盘IO的操作。但是，如果计算过程中涉及数据交换，Spark 也是会把 shuffle 的数据写磁盘的！

3.Spark内置模块

Spark Core：实现了Spark的基本功能，包含任务调度、内存管理、错误恢复、与存储系统交互等模块。Spark Core中还包含了对弹性分布式数据集(Resilient Distributed DataSet，简称RDD)的API定义。

Spark SQL：是Spark用来操作结构化数据的程序包。通过Spark SQL，我们可以使用 SQL或者Apache Hive版本的SQL方言(HQL)来查询数据。Spark SQL支持多种数据源，比如Hive表、Parquet以及JSON等。

Spark Streaming：是Spark提供的对实时数据进行流式计算的组件。提供了用来操作数据流的API，并且与Spark Core中的 RDD API高度对应。

Spark MLlib：提供常见的机器学习(ML)功能的程序库。包括分类、回归、聚类、协同过滤等，还提供了模型评估、数据 导入等额外的支持功能。

集群管理器：Spark 设计为可以高效地在一个计算节点到数千个计算节点之间伸缩计 算。为了实现这样的要求，同时获得最大灵活性，Spark支持在各种集群管理器(Cluster Manager)上运行，包括Hadoop YARN、Apache Mesos，以及Spark自带的一个简易调度 器，叫作独立调度器。

二、安装Spark

1.Spark相关地址

1．官网地址

Apache Spark™ - Unified Engine for large-scale data analytics

2．文档查看地址

Redirecting…

3．下载地址

https://spark.apache.org/downloads.html

https://archive.apache.org/dist/（apache归档）

2.重要角色

2.1 Driver（驱动器）

1、driver进程就是应用的main()函数并且构建sparkContext对象，当我们提交了应用之后，便会启动一个对应的driver进程，driver本身会根据我们设置的参数占有一定的资源（主要指cpu core和memory）。 2、driver可以运行在master上，也可以运行worker上（根据部署模式的不同）。 3、driver首先会向集群管理者（standalone、yarn，mesos）申请spark应用所需的资源，也就是executor，然后集群管理者会根据spark应用所设置的参数在各个worker上分配一定数量的executor，每个executor都占用一定数量的cpu和memory。在申请到应用所需的资源以后，driver就开始调度和执行我们编写的应用代码了。 4、driver进程会将我们编写的spark应用代码拆分成多个stage，每个stage执行一部分代码片段，并为每个stage创建一批task，然后将这些tasks分配到executor中执行。

2.2 Executor（执行器）

executor进程宿主在worker节点上，一个worker可以有多个executor。每个executor持有一个线程池，每个线程可以执行一个task，executor执行完task以后将结果返回给driver，每个executor执行的task都属于同一个应用。 此外executor还有一个功能就是为应用程序中要求缓存的RDD提供内存式存储，RDD是直接缓存在executor进程内的，因此任务可以在运行时充分利用缓存数据加速运算。

Standalone伪分布式模式

Standalone: spark的代码只需要spark自己的内置调度器，不需要借助第三方资源调度器(yarn)

资源调度器：yarn、mesos、k8s

为什么不用spark自带的standalone，要使用yarn呢?

准备工作：正常安装JDK、Hadoop(启动hdfs)

1. 上传并解压spark安装包、配置环境变量

[root@hadoop10 modules]# tar -zxf spark-3.1.2-bin-hadoop3.2.tgz -C /opt/installs [root@hadoop10 installs]# mv spark-3.1.2-bin-hadoop3.2 spark vi /etc/profile export SPARK_HOME=/opt/installs/spark export PATH=$PATH:$SPARK_HOME/bin

1. 修改配置文件

[root@hadoop10 installs]# cd spark/conf [root@hadoop10 conf]# mv workers.template workers [root@hadoop10 conf]# mv spark-env.sh.template spark-env.sh [root@hadoop10 conf]# vi workers #配置Spark集群节点主机名，在该主机上启动worker进程, 删除localhost hadoop10 [root@hadoop10 conf]# vi spark-env.sh #增加 JAVA_HOME 配置 export JAVA_HOME=/opt/installs/jdk1.8 #声明Spark集群中Master的主机名和端口号 export SPARK_MASTER_HOST=hadoop10 # 主节点内部通讯端口，用于接收客户端请求 export SPARK_MASTER_PORT=7077 # 主节点用于供外部提供浏览器web访问的端口 export SPARK_MASTER_WEBUI_PORT=8080 export SPARK_WORKER_WEBUI_PORT=8081

1. 安装Anaconda

1. 上传 将Anaconda3-2021.05-Linux-x86_64.sh上传到/opt/modules目录下 

2. 安装 # 添加执行权限 chmod u+x Anaconda3-2021.05-Linux-x86_64.sh # 执行 sh ./Anaconda3-2021.05-Linux-x86_64.sh # 过程 #第一次：【直接回车，然后按q】 Please, press ENTER to continue >>> #第二次：【输入yes】 Do you accept the license terms? [yes|no] [no] >>> yes #第三次：【输入解压路径：/opt/installs/anaconda3】 [/root/anaconda3] >>> /opt/installs/anaconda3 #第四次：【输入yes，是否在用户的.bashrc文件中初始化 Anaconda3的相关内容】 Do you wish the installer to initialize Anaconda3 by running conda init? [yes|no] [no] >>> yes 

3. 刷新 在/root/.bashrc中增加 export PYSPARK_PYTHON=/opt/installs/anaconda3/bin/python3 # 刷新环境变量 source /root/.bashrc # 激活虚拟环境，如果需要关闭就使用：conda deactivate conda activate

补充

#创建环境：创建一个新的 Python 环境，可以指定 Python 版本和需要的包。 conda create -n myenv python=3.8 <package_name> #这将创建一个名为 myenv 的环境，并在其中安装 Python 3.8 和指定的包。 #激活环境：激活一个已创建的环境。 conda activate myenv #离开环境：离开当前激活的环境。 conda deactivate #列出已安装的包：列出当前环境中已安装的所有包。 #安装包：安装指定的包。 conda list、conda install、conda search

conda config --set auto_activate_base false

1. 启动spark

[root@hadoop10 spark]# /opt/installs/spark/sbin/start-all.sh [root@hadoop10 spark]# jps 2054 Jps 2008 Worker 1933 Master

方式1: Spark Python Shell (基于命令行的方式 主要用于平时测试)

新建一个wc.txt如下，并且上传到hdfs之上 hello hello hello world world hello [root@hadoop10 spark]# pyspark --master spark://hadoop10:7077 >>> sc.textFile("hdfs://hadoop10:8020/wc.txt").flatMap(lambda x:x.split(" ")).map(lambda x:(x,1)).reduceByKey(lambda x,y:x+y).saveAsTextFile("hdfs://hadoop10:8020/spark/out")

方式2: spark-submit (作业提交的方式)

[root@hadoop10 spark]# spark-submit --master spark://hadoop10:7077 /opt/installs/spark/examples/src/main/python/pi.py 10

在服务器运行pyspark 命令的时候，可以通过浏览器访问http://hadoop10:4040页面，监控spark中job的运行情况，但是一旦退出pyspark命令，则4040界面无法访问

JobHistoryServer配置

1. 修改spark-defaults.conf.template名称, 修改spark-defaults.conf文件，开启Log

[root@hadoop10 conf]# mv spark-defaults.conf.template spark-defaults.conf [root@hadoop10 conf]# vi spark-defaults.conf spark.eventLog.enabled true spark.eventLog.dir hdfs://hadoop10:8020/spark-logs

注意：HDFS上的目录需要提前存在。

[root@hadoop10 conf]# hdfs dfs -mkdir /spark-logs

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

[root@hadoop10 conf]# vi spark-env.sh SPARK_HISTORY_OPTS="-Dspark.history.fs.logDirectory=hdfs://hadoop10:8020/spark-logs"

1. 启动对应的服务

[root@hadoop10 spark]# /opt/installs/spark/sbin/start-history-server.sh 

[root@hadoop10 spark]# jps 9645 HistoryServer # 对应启动的进程名称

1. 查看历史服务http://hadoop10:18080