20220311

參數(shù)調(diào)節(jié)
把executor數(shù)量調(diào)小,其他參數(shù)值調(diào)大，不容易報(bào)錯(cuò)

一.指定spark executor 數(shù)量的公式
executor 數(shù)量 = spark.cores.max/spark.executor.cores
spark.cores.max 是指你的spark程序需要的總核數(shù)
spark.executor.cores 是指每個(gè)executor需要的核數(shù)
二.指定并行的task數(shù)量
spark.default.parallelism
參數(shù)說(shuō)明：該參數(shù)用于設(shè)置每個(gè)stage的默認(rèn)task數(shù)量。這個(gè)參數(shù)極為重要，如果不設(shè)置可能會(huì)直接影響你的Spark作業(yè)性能。
參數(shù)調(diào)優(yōu)建議：Spark作業(yè)的默認(rèn)task數(shù)量為500~1000個(gè)較為合適。很多同學(xué)常犯的一個(gè)錯(cuò)誤就是不去設(shè)置這個(gè)參數(shù)，那么此時(shí)就會(huì)導(dǎo)致Spark自己根據(jù)底層HDFS的block數(shù)量來(lái)設(shè)置task的數(shù)量，默認(rèn)是一個(gè)HDFS block對(duì)應(yīng)一個(gè)task。通常來(lái)說(shuō)，Spark默認(rèn)設(shè)置的數(shù)量是偏少的（比如就幾十個(gè)task），如果task數(shù)量偏少的話(huà)，就會(huì)導(dǎo)致你前面設(shè)置好的Executor的參數(shù)都前功盡棄。試想一下，無(wú)論你的Executor進(jìn)程有多少個(gè)，內(nèi)存和CPU有多大，但是task只有1個(gè)或者10個(gè)，那么90%的Executor進(jìn)程可能根本就沒(méi)有task執(zhí)行，也就是白白浪費(fèi)了資源！因此Spark官網(wǎng)建議的設(shè)置原則是，設(shè)置該參數(shù)為num-executors * executor-cores的2~3倍較為合適，比如Executor的總CPU core數(shù)量為300個(gè)，那么設(shè)置1000個(gè)task是可以的，此時(shí)可以充分地利用Spark集群的資源。
三. 命令示例
1
spark-submit --class com.cjh.test.WordCount --conf spark.default.parallelism=12 --conf spark.executor.memory=800m --conf spark.executor.cores=2 --conf spark.cores.max=6 my.jar
四.其他調(diào)優(yōu)參數(shù)
spark.storage.memoryFraction
參數(shù)說(shuō)明：該參數(shù)用于設(shè)置RDD持久化數(shù)據(jù)在Executor內(nèi)存中能占的比例，默認(rèn)是0.6。也就是說(shuō)，默認(rèn)Executor 60%的內(nèi)存，可以用來(lái)保存持久化的RDD數(shù)據(jù)。根據(jù)你選擇的不同的持久化策略，如果內(nèi)存不夠時(shí)，可能數(shù)據(jù)就不會(huì)持久化，或者數(shù)據(jù)會(huì)寫(xiě)入磁盤(pán)。
參數(shù)調(diào)優(yōu)建議：如果Spark作業(yè)中，有較多的RDD持久化操作，該參數(shù)的值可以適當(dāng)提高一些，保證持久化的數(shù)據(jù)能夠容納在內(nèi)存中。避免內(nèi)存不夠緩存所有的數(shù)據(jù)，導(dǎo)致數(shù)據(jù)只能寫(xiě)入磁盤(pán)中，降低了性能。但是如果Spark作業(yè)中的shuffle類(lèi)操作比較多，而持久化操作比較少，那么這個(gè)參數(shù)的值適當(dāng)降低一些比較合適。此外，如果發(fā)現(xiàn)作業(yè)由于頻繁的gc導(dǎo)致運(yùn)行緩慢（通過(guò)spark web ui可以觀(guān)察到作業(yè)的gc耗時(shí)），意味著task執(zhí)行用戶(hù)代碼的內(nèi)存不夠用，那么同樣建議調(diào)低這個(gè)參數(shù)的值。
spark.shuffle.memoryFraction
參數(shù)說(shuō)明：該參數(shù)用于設(shè)置shuffle過(guò)程中一個(gè)task拉取到上個(gè)stage的task的輸出后，進(jìn)行聚合操作時(shí)能夠使用的Executor內(nèi)存的比例，默認(rèn)是0.2。也就是說(shuō)，Executor默認(rèn)只有20%的內(nèi)存用來(lái)進(jìn)行該操作。shuffle操作在進(jìn)行聚合時(shí)，如果發(fā)現(xiàn)使用的內(nèi)存超出了這個(gè)20%的限制，那么多余的數(shù)據(jù)就會(huì)溢寫(xiě)到磁盤(pán)文件中去，此時(shí)就會(huì)極大地降低性能。
參數(shù)調(diào)優(yōu)建議：如果Spark作業(yè)中的RDD持久化操作較少，shuffle操作較多時(shí)，建議降低持久化操作的內(nèi)存占比，提高shuffle操作的內(nèi)存占比比例，避免shuffle過(guò)程中數(shù)據(jù)過(guò)多時(shí)內(nèi)存不夠用，必須溢寫(xiě)到磁盤(pán)上，降低了性能。此外，如果發(fā)現(xiàn)作業(yè)由于頻繁的gc導(dǎo)致運(yùn)行緩慢，意味著task執(zhí)行用戶(hù)代碼的內(nèi)存不夠用，那么同樣建議調(diào)低這個(gè)參數(shù)的值。

spark.spark.default.parallelism與sql.shuffle.partitions的設(shè)置:

Spark中RDD對(duì)應(yīng)有partition的概念，每個(gè)partition都會(huì)對(duì)應(yīng)一個(gè)task，task越多，在處理大規(guī)模數(shù)據(jù)的時(shí)候，就會(huì)越有效率。但是并不是task越多越好，如果平時(shí)測(cè)試，或者數(shù)據(jù)量沒(méi)有那么大，則沒(méi)有必要task數(shù)量太多。

我的第一個(gè)query程序，有200個(gè)task，我改成了50個(gè)，節(jié)約了1s左右。
參數(shù)可以通過(guò)spark_home/conf/spark-default.conf配置文件設(shè)置:
spark.sql.shuffle.partitions 50
spark.default.parallelism 10
上邊兩個(gè)參數(shù)，第一個(gè)是針對(duì)spark sql的task數(shù)量，我的程序邏輯是將rdd首先轉(zhuǎn)換成dataframe，然后進(jìn)行query，所以對(duì)應(yīng)第一個(gè)參數(shù)。
而如果程序是非sql則第二個(gè)參數(shù)生效。

SparkContext中默認(rèn)有defaultMinPartitions指定最小的分區(qū)數(shù)；
def defaultMinPartitions: Int = math.min(defaultParallelism, 2)
SparkContext中生成RDD的接口中往往指定的是minPartitions最小分區(qū)數(shù)目。

https://www.cnblogs.com/hadoop-dev/p/6669232.html
spark分區(qū)數(shù),task數(shù)目,core數(shù),worker節(jié)點(diǎn)個(gè)數(shù),excutor數(shù)量梳理

作者：王燚光
https://www.zhihu.com/question/33270495/answer/934

輸入可能以多個(gè)文件的形式存儲(chǔ)在HDFS上，每個(gè)File都包含了很多塊，稱(chēng)為Block。
當(dāng)Spark讀取這些文件作為輸入時(shí)，會(huì)根據(jù)具體數(shù)據(jù)格式對(duì)應(yīng)的InputFormat進(jìn)行解析，一般是將若干個(gè)Block合并成一個(gè)輸入分片，稱(chēng)為InputSplit，注意InputSplit不能跨越文件。
隨后將為這些輸入分片生成具體的Task。InputSplit與Task是一一對(duì)應(yīng)的關(guān)系。
隨后這些具體的Task每個(gè)都會(huì)被分配到集群上的某個(gè)節(jié)點(diǎn)的某個(gè)Executor去執(zhí)行。

每個(gè)節(jié)點(diǎn)可以起一個(gè)或多個(gè)Executor。
每個(gè)Executor由若干core組成，每個(gè)Executor的每個(gè)core一次只能執(zhí)行一個(gè)Task。
每個(gè)Task執(zhí)行的結(jié)果就是生成了目標(biāo)RDD的一個(gè)partiton。

注意: 這里的core是虛擬的core而不是機(jī)器的物理CPU核，可以理解為就是Executor的一個(gè)工作線(xiàn)程。

而 Task被執(zhí)行的并發(fā)度 = Executor數(shù)目 * 每個(gè)Executor核數(shù)。

至于partition的數(shù)目：

對(duì)于數(shù)據(jù)讀入階段，例如sc.textFile，輸入文件被劃分為多少I(mǎi)nputSplit就會(huì)需要多少初始Task。
在Map階段partition數(shù)目保持不變。
在Reduce階段，RDD的聚合會(huì)觸發(fā)shuffle操作，聚合后的RDD的partition數(shù)目跟具體操作有關(guān)，例如repartition操作會(huì)聚合成指定分區(qū)數(shù)，還有一些算子是可配置的。
RDD在計(jì)算的時(shí)候，每個(gè)分區(qū)都會(huì)起一個(gè)task，所以rdd的分區(qū)數(shù)目決定了總的的task數(shù)目。
申請(qǐng)的計(jì)算節(jié)點(diǎn)（Executor）數(shù)目和每個(gè)計(jì)算節(jié)點(diǎn)核數(shù)，決定了你同一時(shí)刻可以并行執(zhí)行的task。
比如的RDD有100個(gè)分區(qū)，那么計(jì)算的時(shí)候就會(huì)生成100個(gè)task，你的資源配置為10個(gè)計(jì)算節(jié)點(diǎn)，每個(gè)兩2個(gè)核，同一時(shí)刻可以并行的task數(shù)目為20，計(jì)算這個(gè)RDD就需要5個(gè)輪次。
如果計(jì)算資源不變，你有101個(gè)task的話(huà)，就需要6個(gè)輪次，在最后一輪中，只有一個(gè)task在執(zhí)行，其余核都在空轉(zhuǎn)。
如果資源不變，你的RDD只有2個(gè)分區(qū)，那么同一時(shí)刻只有2個(gè)task運(yùn)行，其余18個(gè)核空轉(zhuǎn)，造成資源浪費(fèi)。這就是在spark調(diào)優(yōu)中，增大RDD分區(qū)數(shù)目，增大任務(wù)并行度的做法。

引用https://www.cnblogs.com/hadoop-dev/p/6669232.html

引用https://blog.csdn.net/wangguohe/article/details/80645978

總結(jié)

以上是生活随笔為你收集整理的pyspark性能调优参数的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。