Yarn模式下Spark安装与配置
Yarn模式下Spark安装与配置
生产环境下基本都在Yarn集群下运行Spark任务，因此本文以Spark 3.1.2为例详细说明Spark集群的安装与配置步骤。
注：本文搭建Spark集群是在已正确安装Hadoop集群之上，存放在/opt/module/hadoop-3.1.3路径下。
安装包下载地址
官网地址：http://spark.apache.org/
下载地址：https://archive.apache.org/dist/spark/
从Spark官网下载jar包地址
安装使用
1、 上传并解压解压spark-3.1.2-bin-hadoop3.2.tgz
[user_name@hadoop100 module]$ tar -zxvf spark-3.1.2-bin-hadoop3.2.tgz -C /opt/module/
[user_name@hadoop100 module]$ mv /opt/module/spark-3.1.2-bin-hadoop3.2 /opt/module/spark-3.1.2
2、配置SPARK_HOME环境变量
[user_name@hadoop100 module]$ sudo vim /etc/profile.d/my_env.sh
添加如下内容：
# SPARK_HOME
export SPARK_HOME=/opt/module/spark-3.1.2
export PATH=$PATH:$SPARK_HOME/bin
source my_env.sh 使其生效
3、修改/opt/module/spark-3.1.2/conf/spark-env.sh，添加HADOOP_CONF_DIR和YARN_CONF_DIR配置，保证spark可以利用YARN进行资源管理和利用HDFS进行数据存储。
[user_name@hadoop100 conf]$ mv spark-env.sh.template spark-env.sh
[user_name@hadoop100 conf]$ vim spark-env.sh
export HADOOP_CONF_DIR=/opt/module/hadoop-3.1.3/etc/hadoop
export YARN_CONF_DIR=/opt/module/hadoop-3.1.3/etc/hadoop
配置依赖的spark jar包
Spark On Yarn不需要搭建Spark集群，只需要在提交作业的节点上安装单机版spark，并同时将依赖的jar包事先传入到hdfs上即可。
原理：① excutor会将运行task所需的jar包从从HDFS下载到本地磁盘的一个缓存目录中“资源本地化”；
② 执行task时会将本地缓存的JARs添加到它们的类路径（Classpath）中，这样即使该节点没有安装Spark，JVM进程内部也拥有了运行Spark程序所需的所有类。
配置步骤：
1、创建一个HDFS目录，并将 $SPARK_HOME/jars/下的所有JAR包上传上去。
hdfs dfs -mkdir -p /spark-3.1.2/jars
hdfs dfs -put /jars/* /spark-3.1.2/jars/
2、修改spark-default.conf文件，配置jar包所在的hdfs路径。
mv spark-defaults.conf.template spark-defaults.conf
vim spark-defaults.conf
spark.yarn.jars hdfs://hadoop100:8020/spark-3.1.2/jars/*
注：即使其他节点安装了Spark，如果 没有配置spark.yarn.jars、spark.executor.install.dir（本地spark安装目录）参数以及提交任务时没有指定该参数，YARN 会退回到一种传统的、效率较低的模式，NodeManager 会尝试从提交应用的客户端机器拉取 Spark 库。
配置Spark历史服务器
需要针对Yarn模式，再次配置一下历史服务器。
1、修改spark-default.conf文件，配置日志存储路径，指定 Spark 将日志输入到 HDFS 中。
[user_name@hadoop100 conf]$ vim spark-defaults.conf
spark.eventLog.enabled true
spark.eventLog.dir hdfs://hadoop100:8020/spark-history
2、在文件末尾添加HistoryServer 启动参数, 使得 HistoryServer 在启动的时候读取 HDFS 中写入的 Spark 日志。
spark.history.fs.logDirectory hdfs://hadoop100:8020/spark-history
spark.history.fs.cleaner.enabled true
spark.history.retainedApplications 30
3、为 Spark 创建 HDFS 中的日志目录
hdfs dfs -mkdir - p /spark-history
关联Spark历史日志
为了能从Yarn上关联到Spark历史服务器，需要配置spark历史服务器关联路径。
目的：点击yarn（8088）上spark任务的history按钮，进入的是spark历史服务器（18080），而不再是yarn历史服务器（19888）。
1、修改配置文件/opt/module/spark-3.1.2/conf/spark-defaults.conf，添加如下内容：
spark.history.ui.port 18080
spark.yarn.historyServer.address hadoop100:18080
修改日志级别(可选)
Spark的日志默认是INFO级别，非常详细，如果不想看这么多可以修日志级别为WARN级别。
① 临时调整。在提交任务时设置 --conf "spark.log.level=WARN"
② 永久修改
mv log4j.properties.template log4j.properties
vim log4j.properties
综上，Spark集群配置完毕。为保证每台机器都可以提交Spark应用，分发Spark文件到所有集群节点。
Spark作业提交
启动组件
1、启动hadoop集群，包含hdfs、yarn以及历史服务器；
2、启动Spark历史服务
[user_name@hadoop100 spark-3.1.2]$ sbin/start-history-server.sh
成功启动服务后，可以看到HistoryServer进程。
提交Spark 任务
在实际生产中，常使用spark-submit命令向集群中提交jar任务，命令如下：
./bin/spark-submit \
--master <master-url> \
--deploy-mode <deploy-mode> \
--conf <key>=<value> \
... # other options
--class <main-class> \
<application-jar> \
[application-arguments]
① --master：集群的 Master URL，决定Spark运行模式。
| Master URL | 含义 |
|---|---|
local | 使用一个工作线程本地化运行 Spark（完全不并行）。 |
local[K] | 使用 K 个 工作线程本地返还运行 Spark。（理想情况下，将其设置为机器上的内核数）。 |
local[K,F] | 使用 K 个工作线程和 F 个 maxFailures 在本地运行 Spark |
local[*] | 在本地运行 Spark，工作线程与机器上的逻辑核心一样多。 |
local[*,F] | 在本地运行 Spark，工作线程数与您机器上的逻辑核心数和 F maxFailures 数一样多。 |
local-cluster[N,C,M] | 本地集群模式仅用于单元测试。它在单个 JVM 中模拟分布式集群，具有 N 个工作器，每个工作器 C 核和每个工作器 M MiB 内存。 |
spark://HOST:PORT | Standalone 模式 |
spark://HOST1:PORT1,HOST2:PORT2 | Standalone 高可用模式 |
mesos://HOST:PORT | Mesos 模式 |
yarn | YARN 模式 |
k8s://HOST:PORT | Kubernetes 模式 |
② --deploy-mode：是否将 driver 部署在工作节点(cluster)或本地的外部客户端(client) (默认值：client)。
③ --conf：key=value 格式的 Spark 属性配置。
④ application-jar：提交要运行jar包的路径，如本地路径或hdfs路径
⑤ --class：应用程序的入口（例如 org.apache.spark.examples.SparkPi）。
⑥ application-arguments：传递给入口类中 main 方法的参数。
运行官方实例
以官方提供的SparkPi任务为例，向YARN集群提交运行指令：
/opt/module/spark-3.1.2/bin/spark-submit \
--master yarn \
--deploy-mode client \
--class org.apache.spark.examples.SparkPi \
/opt/module/spark-3.1.2/examples/jars/spark-examples_2.12-3.1.2.jar \
10
如果日志等级是INFO，从控制台输出日志可看到：① 会将要运行的spark-examples_2.12-3.1.2.jar包提交到hdfs上，供其他 worker 节点会从这里下载要运行的 JAR 文件；② 由于任务运行所依赖的spark jar包已经事先上传至hdfs上，就会跳过复制操作，直接使用现有文件。
此外，在运行任务时，查看集群各节点的后天进程，可以看到启动了Executor。
运行IDEA开发独立应用
在实际生产中，更多的是根据特定业务逻辑自行编写程序后打jar包再提交运行，具体步骤参考：IDEA开发Spark应用
查看日志
成功运行完任务后，打开Web页面：http://hadoop102:8088/cluster/，查看执行日志
点击“history”跳转到Spark UI界面：http://hadoop100:18080/
native-hadoop library报错处理
在提交任务时可能会报如下错误，这是由于Spark 无法加载本地 Hadoop 库。
WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
原因可能为：
① 缺少本地 Hadoop 库：系统没有安装或正确配置 Hadoop 的本地库；
② 环境变量未设置：HADOOP_HOME或相关环境变量未正确配置。
验证步骤：
① 检查 Hadoop 本地库是否存在：
# 检查 Hadoop 安装目录中的 native 库
find $HADOOP_HOME -name "libhadoop.so*" -type f
find $HADOOP_HOME -name "*.so" | grep native
如果有以下类似文件输出说明本地库存在。
lib/native/libhadoop.so
lib/native/libhdfs.so
lib/native/libnativetask.so
② 验证环境变量是否设置：
echo $HADOOP_HOME
echo $LD_LIBRARY_PATH
如果未输出，需在 spark-env.sh中添加：
export HADOOP_HOME=/opt/module/hadoop-3.1.3
export HADOOP_CONF_DIR=$HADOOP_HOME/etc/hadoop
export LD_LIBRARY_PATH=$HADOOP_HOME/lib/native:$LD_LIBRARY_PATH
export SPARK_DIST_CLASSPATH=$($HADOOP_HOME/bin/hadoop classpath)
配置参数生效优先级
Spark 的配置优先级从高到低为：
① 提交命令行参数 (–master, --conf)
② SparkConf.set() 方法
③ 配置文件 (spark-defaults.conf)
④ 系统属性 (System.getProperties())
⑤ 默认值
Spark on Hive配置
在生产中，Hive 通常是作为 Spark 的数据源，并通过 Hive 的 MetaStore 获取到 Hive 表的元数据运行 Spark SQL (Spark On Hive)，下面介绍一下配置实现Spark On Hive。
配置 Spark On Hive
注：要求已经提前安装好Hive了
1、将 $HIVE_HOME/conf 目录下的 hive-site.xml 文件复制到 $SPARK_HOME/conf 目录下；
2、将 $HIVE_HOME/lib 目录下的 MySQL 驱动文件复制到 $SPARK_HOME/jars 目录下；
3、分发上述配置到其他节点
测试配置是否成功
1、启动hadoop集群，包含hdfs、yarn以及历史服务器
2、启动hive
① 在hive运行的服务器上，首先启动metastore服务，然后启动hiveserver2服务。注： hiveserver2服务启动之后需要稍等一会才可以对外提供服务，如果立马启动beeline连接hiverserver2，则会报错连接不上，因此要等等！！！
nohup /opt/module/hive-3.1.2/bin/hive --service metastore &
nohup /opt/module/hive-3.1.2/bin/hive --service hiveserver2 &
② 利用Beeline客户端（JDBC的客户端），通过JDBC协议和Hiveserver2服务进行通信
beeline
! connect jdbc:hive2://hadoop100:10000
③ 输入账户名和密码。
3、启动 Spark SQL CLI 测试
spark-sql --master yarn --deploy-mode client
4、查询hive表。如果显示在hive中的数据库、表以及可以正常查询出表数据说明配置成功。
5、进一步可以使用 Hive 客户端对比查询结果。
IDEA配置执行Spark On Hive
1、引入基本依赖
<dependencies>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-client</artifactId>
<version>3.1.3</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.12</artifactId>
<version>3.1.2</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-hive_2.12</artifactId>
<version>3.1.2</version>
</dependency>
</dependencies>
注：如果会报错java.lang.ClassNotFoundException: com.fasterxml.jackson.annotation.JsonMerge，这是因为Spark集群环境中缺少Jackson JSON库的特定版本，需要添加依赖。
<properties>
<jackson.version>2.15.2</jackson.version>
</properties>
<dependencies>
<!-- Jackson Core -->
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-core</artifactId>
<version>${jackson.version}</version>
</dependency>
<!-- Jackson Annotations -->
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-annotations</artifactId>
<version>${jackson.version}</version>
</dependency>
<!-- Jackson Databind -->
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
<version>${jackson.version}</version>
</dependency>
<!-- 如果使用Scala，可能还需要jackson-module-scala -->
<dependency>
<groupId>com.fasterxml.jackson.module</groupId>
<artifactId>jackson-module-scala_2.12</artifactId>
<version>${jackson.version}</version>
</dependency>
</dependencies>
2、配置文件
把hadoop集群和hive的配置文件 core-site.xml，hdfs-site.xml，hive-site.xml 拷贝到 resource目录下。
3、创建测试代码，执行如果查询到数据说明配置成功
object SparkSQLDemo{
def main(args: Array[String]): Unit = {
val spark: SparkSession = SparkSession.builder()
.appName(s"${SparkSQLDemo.getClass.getSimpleName}")
.master("local[*]")
.enableHiveSupport() //开启对 Hive 的支持
.getOrCreate()
// 执行查询
spark.sql("use hive_test")
spark.sql("select * from user_table limit 2").show()
}
}
更多推荐
23
0- 0
已为社区贡献12条内容
所有评论(0)