PySpark Yarn集群模式

目录

简介

一、PySpark简介

二、YARN模式概述

三、配置环境

1. 安装与配置Spark

2. 配置Hadoop和YARN

3. 启动yarn

四、编写PySpark脚本

五、提交PySpark作业到YARN

参数解释：

六、常见问题及解决

七、总结

---

简介

随着大数据的普及，Spark作为主流的分布式计算框架，在大数据处理中扮演着重要角色。YARN（Yet Another Resource Negotiator）作为资源调度和管理框架，与Spark结合后可以更好地管理和调度资源。本篇文章将详细介绍如何在YARN集群模式下运行PySpark应用程序，适合有一定Spark基础的开发者。

一、PySpark简介

PySpark是Spark的Python API，支持使用Python编写代码并提交到Spark集群运行。在大规模数据处理中，PySpark可以利用Python的简单性和Spark的分布式处理能力，极大地提升数据处理效率。

二、YARN模式概述

YARN模式是Spark常用的集群模式之一。YARN可以有效地调度集群资源，并提供容错能力。通过将Spark应用程序提交到YARN，用户可以在Hadoop集群上更高效地执行计算任务。

Spark在YARN上运行时有两种模式：

1. Cluster模式：Driver运行在YARN集群中，适用于生产环境。
2. Client模式：Driver运行在提交应用程序的客户端，适用于调试和开发。

三、配置环境

在PySpark集群模式下运行需要配置Spark和YARN环境。

1. 安装与配置Spark

下载并解压Spark后，设置环境变量SPARK_HOME，并将其添加到PATH中。

cd /opt/modules/
tar -zxf spark-3.1.2-bin-hadoop3.2.tgz -C /opt/installs
cd /opt/installs
mv spark-3.1.2-bin-hadoop3.2 spark-yarn
rm -rf /opt/installs/spark
ln -s /opt/installs/spark-yarn /opt/installs/spark

修改$SPARK_HOME/conf/spark-env.sh文件，指定Hadoop和YARN相关配置：

cd /opt/installs/spark/conf
mv spark-env.sh.template spark-env.sh
vim /opt/installs/spark/conf/spark-env.sh
## 22行左右设置JAVA安装目录、HADOOP和YARN配置文件
目录
export JAVA_HOME=/opt/installs/jdk
export HADOOP_CONF_DIR=/opt/installs/hadoop/etc/hadoop
export YARN_CONF_DIR=/opt/installs/hadoop/etc/hadoop
## 历史日志服务器
export SPARK_DAEMON_MEMORY=1g
export SPARK_HISTORY_OPTS="-Dspark.history.fs.logDirectory=hdfs://bigdata01:9820/spark/eventLogs/ -Dspark.history.fs.cleaner.enabled=true"

修改$SPARK_HOME/conf/spark-defaults.conf文件:

cd /opt/installs/spark/conf
mv spark-defaults.conf.template spark-defaults.confvim spark-defaults.conf
## 添加内容：
spark.eventLog.enabled           true
spark.eventLog.dir           hdfs://bigdata01:9820/spark/eventLogs
spark.eventLog.compress           true
spark.yarn.historyServer.address bigdata01:18080
spark.yarn.jars           hdfs://bigdata01:9820/spark/jars/*

修改$SPARK_HOME/conf/log4j.properties文件:

mv log4j.properties.template log4j.properties# 修改级别为WARN,打印日志少一点。

上传spark jar包:

# 因为YARN中运行Spark，需要用到Spark的一些类和方法
# 如果不上传到HDFS，每次运行YARN都要上传一次，比较慢
# 所以自己手动上传一次，以后每次YARN直接读取即可
hdfs dfs -mkdir -p /spark/jars/
hdfs dfs -put /opt/installs/spark/jars/* /spark/jars/

2. 配置Hadoop和YARN

确保Hadoop集群已配置并正常运行，core-site.xml和yarn-site.xml应配置正确。

 修改$SPARK_HOME/conf/yarn-site.xml文件:

cd /opt/installs/hadoop/etc/hadoop# 检查以下内置少什么，就配什么。
<property><name>yarn.log-aggregation-enable</name><value>true</value>
</property><!-- 历史日志在HDFS保存的时间，单位是秒 -->
<!-- 默认的是-1，表示永久保存 -->
<property><name>yarn.log-aggregation.retain-seconds</name><value>604800</value>
</property><property><name>yarn.log.server.url</name><value>http://bigdata01:19888/jobhistory/logs</value>
</property><!-- 关闭yarn内存检查 -->
<property><name>yarn.nodemanager.pmem-check-enabled</name><value>false</value>
</property>
<property><name>yarn.nodemanager.vmem-check-enabled</name><value>false</value>
</property>

 将Master节点的文件分发到Worker节点(xsync.sh)：

xsync.sh yarn-site.xml

xsync.sh /opt/installs/spark-yarn
# 超链接也分发一下：
xsync.sh /opt/installs/spark

3. 启动yarn

start-yarn.sh
# 启动MR的JobHistoryServer：19888
mapred --daemon start historyserver
# 启动Spark的HistoryServer:18080
/opt/installs/spark/sbin/start-history-server.sh

四、编写PySpark脚本

下面是一个简单的PySpark示例代码，将在YARN集群模式下运行。

from pyspark import SparkContext, SparkConf from pyspark.sql import SparkSession
# 创建Spark配置
conf = SparkConf().setAppName("PySpark YARN Example")
# 创建SparkSession spark = SparkSession.builder.config(conf=conf).getOrCreate()
# 示例操作：读取文件并统计单词数量
data = spark.read.text("hdfs:///path/to/input.txt") words = data.rdd.flatMap(lambda line: line.value.split(" ")) word_counts = words.map(lambda word: (word, 1)).reduceByKey(lambda a, b: a + b) word_counts.collect()
spark.stop()

五、提交PySpark作业到YARN

在集群模式下，将PySpark脚本提交到YARN中运行，使用以下命令：

spark-submit \
--master yarn \
--deploy-mode cluster \
--driver-memory 4G \
--executor-memory 2G \
--executor-cores 2 \
/path/to/your_script.py

参数解释：

• --master yarn：指定YARN作为集群管理器。
• --deploy-mode cluster：集群模式，Driver运行在YARN上。
• --driver-memory：Driver分配的内存。
• --executor-memory：每个Executor分配的内存。
• --executor-cores：每个Executor分配的CPU核心数。

六、常见问题及解决

1. 网络超时问题：在YARN模式下，网络超时是一个常见问题，尤其在集群模式中。可以增加spark.network.timeout的时间来解决。

2. 内存不足：增加driver-memory或executor-memory的分配量。

3. 依赖包丢失：可以使用--py-files参数将Python依赖包提交到集群。

七、总结

在YARN集群模式下运行PySpark作业可以充分利用资源管理器的调度能力，适用于生产环境。通过配置Spark、Hadoop和YARN，编写PySpark脚本并提交到YARN，用户可以高效地执行分布式计算任务。