Spark基础-任务提交相关参数

整理一下用过的spark相关的参数

spark应用提交命令spark-submit的常用参数（使用spark-submit --help可以查看所有参数， 有一些参数在下面的spark配置属性定义了，也没有额外列出）

参数	默认值	含义
--master	local[*]	spark集群的master url，可以是yarn, local等值(master url取值列表 )
--deploy-mode	client	有cluster和client两种模式，决定driver是在worker节点上，还是在本地作为一个外部client。
--name	/	应用的名称
--conf	/	额外的spark配置属性，以key=value的形式表示
--py-files	/	用逗号分隔的.zip, .egg, .py文件，将其路径防止在PYTHONPATH给python应用使用

常用spark配置属性

参数	默认大小	含义	官方文档对应类别
spark.driver.memory	1g	driver内存，在client模式下必须通过spark-submit的 --driver-memory来设置，而不能通过SparkConf来设置	Application Properties
spark.driver.cores	1	driver对应的核数，只有在cluster模式下可以设置	Application Properties
spark.driver.memoryOverhead	driverMemory * spark.driver.memoryOverheadFactor, 最小值为384M	在cluster模式下driver被分配的non-heap 内存。这块内存是用于虚拟机的开销、内部的字符串、还有一些本地开销(比如python需要用到的内存)等。当spark.memory.offHeap.enabled=true时，非堆内存包括堆外内存和其他driver进程使用的内存（例如与PySpark driver一起使用的python进程）和其他在同一个容器中运行的非driver进程使用的内存 。 所以运行driver的容器的最大内存大小由spark.driver.memoryOverhead和spark.driver.memory之和确定。	Application Properties
spark.driver.memoryOverheadFactor	0.1	driver 内存被分配为non-heap内存的比例，如果出现了"Memory Overhead Exceeded"，调大这个比例有助于预防这个错误。如果spark.driver.memoryOverhead被设置了这个参数就会被忽略。	Application Properties
spark.executor.memory	1g	executor的内存大小	Application Properties
spark.executor.pyspark.memory	Not set	每个executor被分配给pyspark使用的内存，如果设置了就限制了pyspark的内存上线；如果不设置spark不会限制python的内存使用，取决于应用本身是否会超出与其他non-JVM共享的overhead 内存。	Application Properties
spark.executor.memoryOverhead	executorMemory * spark.executor.memoryOverheadFactor, 最小值为384M	每个executor被分配的额外内存。这块内存是用于虚拟机的开销、内部的字符串、还有一些本地开销(比如python需要用到的内存)等。当spark.executor.pyspark.memory没有配置时，额外内存还包括pyspark的executer内存， 也包括同一个容器中的其他non-executor进程。所以运行executor的容器的最大内存大小由spark.executor.memoryOverhead, spark.executor.memory, spark.memory.offHeap.size ,spark.executor.pyspark.memory之和确定。	Application Properties
spark.executor.memoryOverheadFactor	0.1	executor内存被分配为non-heap内存的比例，如果出现了"Memory Overhead Exceeded"，调大这个比例有助于预防这个错误。如果spark.executor.memoryOverhead被设置了这个参数就会被忽略。	Application Properties
spark.driver.maxResultSize	1g	对于每个spark action(如collect)序列化结果的总大小限制，至少为1M，如果设为0则无限制。如果序列化结果的总大小限制超过这个限制，Job将会中断。将这个值设的很大，可能会造成driver的out-of-memory错误（取决与spark.driver.memory和JVM中对象的overhead内存)，所以选取一个合适的值有助于driver产生out-of-memory错误。	Application Properties
spark.executor.extraJavaOptions	none	传给executor的额外JVM选项，比如GC设置和其他日志。注意不能设置最大堆内存(-Xmx)，最大推内存是通过spark.executor.memory来设置的。当应用出现堆栈溢出的时候，可能可以通过设置如--conf=spark.executor.extraJavaOptions=-Xss50M来解决	Runtime Environment
spark.executor.cores	yarn上为1 standalone模式时为所有可用核数	executor的核数，一个应用的总核数就是num-executors 乘以executor-cores	Execution Behavior
spark.default.parallelism	对于分布式算子如reduceByKey和join，是父RDD里最大partition数，对于像parallelize等没有父RDD的算子，取决于集群模式：Local是机器上的核数；Mesos fine grained为8，其他则是max(2, 所有executor的总核数)	默认的由transformation 算子如 join, reduceByKey, and parallelize 返回的RDD的分区数	Execution Behavior
spark.executor.heartbeatInterval	10s	每个executor与driver之间心跳的间隔。这个值需要比spark.network.timeout小很多	Execution Behavior
spark.memory.fraction	0.6	用来执行和存储的堆内存比例，越小就涉及越频繁的spills和cached data eviction。此配置的目的是为内部元数据、用户数据结构以及稀疏、异常大的数据的不精确大小估计留出内存。推荐使用默认值，如要设置参考调优文档	Memory Management
spark.memory.storageFraction	0.5	不受驱逐的存储内存量，是由spark.memory.fraction预留的区域大小的一部分。 该值越高，可用于执行的工作内存就越少，任务可能会更频繁地溢出到磁盘。推荐使用默认值，如要设置参考调优文档	Memory Management
spark.memory.offHeap.enabled	false	如果设置为true， spark将对某些操作使用off-heap内存，此时需要将spark.memory.offHeap.size设置为正数	Memory Management
spark.memory.offHeap.size	0	off-heap内存，对于堆内存没有影响，如果executor的总内存有硬限制注意缩减JVM堆内存的大小。	Memory Management
spark.network.timeout	120s	所有网络交互的默认超时时间，以下的参数如果没有被设置会用这个参数来代替：spark.storage.blockManagerHeartbeatTimeoutMs, spark.shuffle.io.connectionTimeout, spark.rpc.askTimeout ，spark.rpc.lookupTimeout	networking
spark.shuffle.io.retryWait	5s	(Netty only)重试提取之间等待的时间。重试造成的最大延迟默认为15秒，计算方式为maxRetries * retryWait	shuffle behavior
spark.shuffle.io.maxRetries	3	(Netty only)如果将其设置为非零值，则由于 IO 相关异常而失败的提取将自动重试。在面对长时间 GC 暂停或暂时性网络连接问题时，此重试逻辑有助于稳定大shuffle。	shuffle behavior
spark.sql.broadcastTimeout	300	在广播join中广播等待时间的超时时间(s)	runtime sql configuration
spark.sql.adaptive.enabled	true	当设置为true时，启用自适应查询执行，这会根据运行时的统计信息在查询执行过程中重新优化查询计划。	runtime sql configuration
spark.sql.adaptive.skewJoin.enabled	true	当true且spark.sql.adaptive.enabled=true，spark会在shuffled join中通过切分倾斜的分区来动态的处理数据倾斜	runtime sql configuration
spark.sql.adaptive.coalescePartitions.enabled	true	当true且spark.sql.adaptive.enabled=true，Spark将根据目标大小（由spark.sql.adaptive.advisoryPartitionSizeInBytes指定）合并连续的shuffle分区，以避免太多的小任务	runtime sql configuration
spark.sql.execution.arrow.pyspark.enabled	false	如果为 true，则在 PySpark 中使用 Apache Arrow 进行列式数据传输。优化应用于1.pyspark.sql.DataFrame.toPandas。2. pyspark.sql.SparkSession.createDataFrame 当其输入是 Pandas DataFrame 或 NumPy ndarray. 以下数据类型不支持: TimestampType的ArrayType	runtime sql configuration
spark.sql.shuffle.partitions	200	为join或聚合而shuffle数据时使用的默认分区数	runtime sql configuration
spark.sql.hive.convertMetastoreParquet	true	当设置为 true 时，内置 Parquet 读取器和写入器用于处理使用 HiveQL 语法创建的 Parquet 表，而不是 Hive serde	runtime sql configuration

一个yarn模式下cluster提交，并且使用自定义python环境的例子

spark-submit \
--deploy-mode cluster \
--master yarn \
--driver-memory 4g \
--num-executors 4 \
--executor-memory 2g \
--executor-cores 2 \
--conf spark.sql.broadcastTimeout=36000 \
--conf spark.driver.maxResultSize=1g \
--conf spark.sql.shuffle.partitions=1000 \
--conf spark.yarn.dist.archives=s3a://path/py37-pyarrow.zip#python37 \
--conf spark.yarn.appMasterEnv.PYSPARK_PYTHON=./python37/mypython/bin/python3 \
--conf spark.yarn.appMasterEnv.PYSPARK_DRIVER_PYTHON=./python37/mypython/bin/python3 \
--py-files s3a://path/companymapping.zip \
--queue default \
--name predict_task \
s3a://path/predict.py 20230813

---

参考资料：

1. https://spark.apache.org/docs/latest/configuration.html
2. https://spark.apache.org/docs/latest/submitting-applications.html
3. https://spark.apache.org/docs/latest/running-on-yarn.html#configuration
4. https://zhuanlan.zhihu.com/p/63187650