目录
前言
之前我们已经进行了pyspark环境的搭建以及经过jupyter notebook进行过开发以及实现了一系列的函数功能.
但是一般我们跑spark都是在集群上面跑，只有测试一般在本地上测试，而且每个公司配置的spark集群的端口和设置的参数都有很大出入，故每种情况都有可能发生。所以一般任务提交的参数最好都需要能够清楚的明白对应功能。很多spark任务都会吃大量的内存以及队列资源，合理的安排spark资源十分重要，这些都需要我们在spark-submit指令上面配置。本篇文章将会将spark-submit的各类参数以及参数对应选项和其功能都会列举出来，并且写有对应场景的pyspark任务提交时参数配置，希望能够帮助到大家，还请点赞支持一下博主。
一、PySpark集群运行原理
我在之前的文章中对pyspark集群运行原理有过比较详细的解答：
主节点的Python通过Py4j通信传递SparkContext，最后在JVM Driver上面生成SparkContxt。主节点JVM Driver与其他从节点的JVM Executor通信传输SparkContext，JVM Executor通过分解SparkContext为许多Task，给pyspark.daemon调用pyspark.work 从socket中读取要执行的python函数和数据，开始真正的数据处理逻辑。数据处理完成之后将处理结果写回socket，jvm中通过PythonRDD的read方法读取，并返回结果。最终executor将PythonRDD的执行结果上报到drive上，返回给用户。
总体而言也就是spark和python程序的一种转换，可以理解为将python程序转换成了scala的spark任务去执行，本质上还是spark任务的提交。故配置是和spark任务应该是一致的，从官方文档也能够了解：
二、spark-submit参数详解
在运行spark-submit指令的时候一般还是直接提交shell命令，或是在dolphinscheduler上，总之在集群上都可提交。spark的bin目录中的spark-submit脚本用于在集群上启动应用程序。它可以通过统一的接口使用Spark支持的所有集群管理器，因此不必为每个集群管理器配置应用程序。
对于Python，可以使用spark-submit的--py-files参数来添加.py、.zip或.egg文件，以便随应用程序一起分发。如果依赖多个Python文件，可以将它们打包成.zip或.egg文件。
1.指定运行目录
一旦绑定了用户应用程序，就可以使用bin/spark提交脚本启动它。该脚本负责使用Spark及其依赖项设置类路径，并可以支持Spark支持的不同集群管理器和部署模式：
-
./bin/spark-submit \
-
--class
<main-class> \
-
--master
<master-url> \
-
--deploy-mode
<deploy-mode> \
-
--conf
<key>=
<value> \
-
... # other options
-
<application-jar> \
-
[application-arguments]
其中指定Hadoop配置可以通过HADOOP_CONF_DIR 指定conf目录。
2.--deploy-mode
使用--deploy-mode，指定运行Spark应用程序驱动程序的位置。Spark支持集群和客户端部署模式。
可选参数有两种：
| 选项 | 说明 |
| cluster | 在集群模式下，驱动程序在其中一个工作节点上运行，该节点在应用程序的Spark Web UI上显示为驱动程序。集群模式用于运行生产作业。逻辑运算与任务执行均在cluster上。 |
| client | 在客户端模式下，驱动程序在提交应用程序的本地运行。客户端模式主要用于交互和调试目的。注意，在客户端模式下，只有驱动程序在本地运行，所有其他执行器在集群上的不同节点上运行。这样逻辑运算在client上执行，任务执行在cluster上 |
3.--master
使用--master选项，可以指定要使用哪个集群管理器来运行应用程序。Spark目前支持Yarn、Mesos、Kubernetes、Stand-alone和local。这些的用途如下所述。
| 集群管理器 | 选项参数 | 说明 |
| Yarn | yarn | 提交到yarn模式部署的集群中，前提是集群是使用yarn管理的 |
| Mesos | mesos://HOST:PORT | 提交到mesos模式部署的集群中，并指定主节点的IP与端口 |
| Standalone | spark://HOST:PORT | 提交到standalone模式部署的spark集群中，并指定主节点的IP与端口 |
| Kubernetes | k8s://HOST:PORT k8s://https://HOST:PORT |
提交到Kubernetes模式部署的spark集群中，并指定主节点的IP与端口 |
| local | local local[k] local[K,F] |
local：提交到本地服务器执行，并分配单个线程 local[k]：提交到本地服务器执行，并分配k个线程 使用local[k，F]并指定F及其失败时应运行的尝试次数。 |
如果我们想要提交一个jar包使用，就可以这样写：
-
./bin/spark-submit \
-
--deploy-mode cluster \
-
--master yarn \
-
--class org.apache.spark.examples.SparkPi \
-
/spark-home/examples/jars/spark-examples_versionxx.jar 80
4.驱动程序和执行器资源
在提交应用程序时，还可以指定要为驱动程序和执行程序提供多少内存和内核。
| 选项 | 说明 |
| --driver-memory | Spark驱动程序使用的内存。 |
| --driver-cores | Spark驱动程序使用的CPU内核 |
| --num-executors | 要使用的执行器总数。 |
| --executor-memory | 用于执行器进程的内存量。 |
| --executor-cores | 用于执行器进程的CPU内核数。 |
| --total-executor-cores | 要使用的执行器颜色总数。 |
使用案例仍旧和上述差不多：
-
./bin/spark2-submit \
-
--master yarn \
-
--deploy-mode cluster \
-
--driver-memory 8g \
-
--executor-memory 16g \
-
--executor-cores
2 \
-
--
class
org.
apache.
spark.
examples.
SparkPi \
-
/spark-home/examples/jars/spark-examples_versionxx.
jar
80
5.--files和--verbose
这两个是经常使用到的选项：
| 选项 | 说明 |
| --file | 使用要使用的逗号分隔文件。 通常，这些文件可以是资源文件夹中的文件。 使用此选项，Spark将所有这些文件提交到集群。 |
| --verbose | 显示详细信息。例如，将spark应用程序使用的所有配置写入日志文件。 |
示例：下面的示例使用集群部署模式将应用程序提交给yarn集群管理器，每个执行器具有8g驱动程序内存、16g和2个内核。
-
./bin/spark2-submit \
-
--verbose
-
--master yarn \
-
--deploy-mode cluster \
-
--driver-memory 8g \
-
--executor-memory 16g \
-
--executor-cores
2 \
-
--files /path/log4j.
properties,
/path/file2.
conf,
/path/file3.
json
-
--
class
org.
apache.
spark.
examples.
SparkPi \
-
/spark-home/examples/jars/spark-examples_versionxx.
jar
80
6.Spark提交配置
Spark提交支持使用--config的几种配置，这些配置用于指定应用程序配置、洗牌参数和运行时配置。
大多数这些配置对于用Java、Scala和Python（PySpark）编写的Spark应用程序是相同的
| 配置参数 | 配置说明 |
| spark.sql.shuffle.partitions | 为更广泛的无序转换（联接和聚合）创建的分区数。 |
| spark.executor.memoryOverhead | 集群模式下每个执行器进程要分配的额外内存量，通常是JVM开销的内存。（PySpark不支持） |
| spark.serializer | org.apache.spark.serializer.<br>JavaSerializer（默认） org.apache.spark.serializer.KryoSerializer |
| spark.sql.files.maxPartitionBytes | 读取文件时每个分区使用的最大字节数。默认128MB。 |
| spark.dynamicAllocation.enabled | 指定是根据工作负载动态增加还是减少执行者的数量。默认为true。 |
| spark.dynamicAllocation .minExecutors |
启用动态分配时要使用的最小执行器数。 |
| spark.dynamicAllocation .maxExecutors |
启用动态分配时要使用的最大执行器数。 |
| spark.executor.extraJavaOptions | 指定JVM选项（请参见下面的示例） |
-
-
./bin/spark2-submit \
-
--master yarn \
-
--deploy-mode cluster \
-
--conf
"spark.sql.shuffle.partitions=20000" \
-
--conf
"spark.executor.memoryOverhead=5244" \
-
--conf
"spark.memory.fraction=0.8" \
-
--conf
"spark.memory.storageFraction=0.2" \
-
--conf
"spark.serializer=org.apache.spark.serializer.KryoSerializer" \
-
--conf
"spark.sql.files.maxPartitionBytes=168435456" \
-
--conf
"spark.dynamicAllocation.minExecutors=1" \
-
--conf
"spark.dynamicAllocation.maxExecutors=200" \
-
--conf
"spark.dynamicAllocation.enabled=true" \
-
--conf
"spark.executor.extraJavaOptions=-XX:+PrintGCDetails -XX:+PrintGCTimeStamps" \
-
--files /path/log4j.
properties,
/path/file2.
conf,
/path/file3.
json \
-
--
class
org.
apache.
spark.
examples.
SparkPi \
-
/spark-home/examples/jars/spark-examples_repace-spark-version.
jar
80
或者，也可以将这些全局@$SPARK_HOME/conf/SPARK-defaults.conf设置为适用于每个SPARK应用程序。还可以通过编程方式使用SparkConf进行设置。Pyspark的配置我已经在另一篇文章有写：
PySpark数据分析基础：pyspark.sql.SparkSession类方法详解及操作+代码展示
三.PySpark程序提交配置选项
一般跑这种带有pyspark的python程序会有两种情况：
1.集群中没有安装python环境。
2.集群中没有程序依赖的第三方库。
下面将对于上面这两种情况提供解决方案：
对于已经安装了python环境的话，只要加入--py-file .py文件就好。当想要提交PySpark应用程序时，需要指定要运行的.py文件，并为依赖库指定.egg文件或.zip文件。这里有一下四个参数需要编辑：
| PYSPARK特定配置 | 说明 |
| --py-files | 使用--py文件添加.py、.zip或.egg文件。 |
| --config spark.executor.pyspark.memory | PySpark为每个执行器使用的内存量。 |
| --config spark.pyspark.driver.python | Python二进制可执行文件，用于驱动程序中的PySpark。 |
| --config spark.pyspark.python | Python二进制可执行文件，用于驱动程序和执行器中的PySpark。 |
对于没有加入第三方库采取上述参数直接配置就好了，若是需要加入简单的第三方库（例如jieba库）时，也只需要打包成zip，然后spark-submit命令中动态加载该模块即可。但是如果要使用复杂的第三方库（含.dll，.so文件，即动态加载库，如numpy、pandas），则处理起来就比较特殊，需要将整个python环境都打包，这有点像Pyinstaller。
解决方法如下：
1.构建一套虚拟环境
我们可以通过anaconda生成虚拟环境也可以通过pipenv来打包，总之我们需要包含提交的任务所包含所以依赖的环境，关于pipenv打包我认为是最简单的，大家可以参考这篇文章：
anaconda的打包可以使用可视化UI这样比较便捷，我们可以创建一个新的虚拟环境，只打上我们需要的依赖包，再将整个虚拟Python环境打包为zip，或tar.gz包
进入到虚拟环境下，如/home/hadoop/anaconda3/envs，使用以下命令将虚拟环境进行打包：
zip -r conda_env.zip conda_env # 虚拟环境为conda_env, 打包为conda_env.zip 文件
提交Pyspark Application，通过 --archives 选项指定 zip 包路径（可以将虚拟环境zip上传到HDFS或某个节点下 ）,然后是在cluster、client模式下提交application，示例如下
-
HADOOP_CONF_DIR=/etc/hadoop/conf /data/anaconda3/envs/py3.6.13/bin/spark-submit
-
--master yarn \
-
--queue default \
-
--deploy-mode client \
-
--conf spark.yarn.dist.archives=hdfs:/py3-6-13.zip
#python \
-
--conf spark.yarn.appMasterEnv.PYSPARK_PYTHON=python/py3.6.13/bin/python3.6\
-
--conf spark.pyspark.python=python/py3.6.13/bin/python3.6\
-
--conf spark.pyspark.driver.python=/data/anaconda3/envs/py3.6.13/bin/python3.6\
-
--conf spark.sql.catalogImplementation=hive\
-
--py-files hdfs:/.py.zip ./.py
2. 模块依赖问题原因
1.Driver 负责把本地的资源分发给 Container。
2.在 client 模式下，Driver 就找在 Client 端，可以把 client 端的资源分发，而 cluster 模式下，Driver 可在集群任一节点，该节点是没有资源的。
普通依赖：
在 client 模式下，Driver 可以自动收集资源，分发，无需上传依赖；
在 cluster 模式下，Driver 无法收集资源，必须上传依赖；
复杂依赖：
在工业环境中，使用的多是 cluster 模式，cluster 模式必须上传依赖。
在调试环节，多使用 client 模式，client 只需上传复杂依赖即可。
参阅
Spark Submit Command Explained with Examples
pyspark 中使用集群未安装的python三方库：加载虚拟python环境
转载：https://blog.csdn.net/master_hunter/article/details/128654441