zoukankan      html  css  js  c++  java
  • spark优化项

    一、Shuffle优化项

    1、Shuffle优化配置 - spark.shuffle.file.buffer

    默认值:32k

    参数说明:该参数用于设置shuffle write task的BufferedOutputStream的buffer缓冲大小。将数据写到磁盘文件之前,会先写入buffer缓冲中,待缓冲写满之后,才会溢写到磁盘。

    调优建议:如果作业可用的内存资源较为充足的话,可以适当增加这个参数的大小(比如64k),从而减少shuffle write过程中溢写磁盘文件的次数,也就可以减少磁盘IO次数,进而提升性能。在实践中发现,合理调节该参数,性能会有1%~5%的提升。


    2、Shuffle优化配置 - spark.reducer.maxSizeInFlight

    默认值:48m

    参数说明:该参数用于设置shuffle read task的buffer缓冲大小,而这个buffer缓冲决定了每次能够拉取多少数据。

    调优建议:如果作业可用的内存资源较为充足的话,可以适当增加这个参数的大小(比如96m),从而减少拉取数据的次数,也就可以减少网络传输的次数,进而提升性能。在实践中发现,合理调节该参数,性能会有1%~5%的提升。


    3、Shuffle优化配置 - spark.shuffle.io.maxRetries

    默认值:3

    参数说明:shuffle read task从shuffle write task所在节点拉取属于自己的数据时,如果因为网络异常导致拉取失败,是会自动进行重试的。该参数就代表了可以重试的最大次数。如果在指定次数之内拉取还是没有成功,就可能会导致作业执行失败。

    调优建议:对于那些包含了特别耗时的shuffle操作的作业,建议增加重试最大次数(比如60次),以避免由于JVM的full gc或者网络不稳定等因素导致的数据拉取失败。在实践中发现,对于针对超大数据量(数十亿~上百亿)的shuffle过程,调节该参数可以大幅度提升稳定性。


    4、Shuffle优化配置 - spark.shuffle.io.retryWait

    默认值:5s

    参数说明: shuffle read task从shuffle write task所在节点拉取属于自己的数据时,如果因为网络异常导致拉取失败,是会自动进行重试的,该参数代表了每次重试拉取数据的等待间隔,默认是5s。

    调优建议:建议加大间隔时长(比如60s),以增加shuffle操作的稳定性。


    5、Shuffle优化配置 - spark.shuffle.memoryFraction

    默认值:0.2

    参数说明:该参数代表了Executor内存中,分配给shuffle read task进行聚合操作的内存比例,默认是20%。

    调优建议:在资源参数调优中讲解过这个参数。如果内存充足,而且很少使用持久化操作,建议调高这个比例,给shuffle read的聚合操作更多内存,以避免由于内存不足导致聚合过程中频繁读写磁盘。在实践中发现,合理调节该参数可以将性能提升10%左右。


    6、Shuffle优化配置 - spark.shuffle.manager

    默认值:sort

    参数说明:该参数用于设置ShuffleManager的类型。Spark 1.5以后,有三个可选项:hash、sort和tungsten-sort。HashShuffleManager是Spark 1.2以前的默认选项,但是Spark 1.2以及之后的版本默认都是SortShuffleManager了。tungsten-sort与sort类似,但是使用了tungsten计划中的堆外内存管理机制,内存使用效率更高。

    调优建议:由于SortShuffleManager默认会对数据进行排序,因此如果你的业务逻辑中需要该排序机制的话,则使用默认的SortShuffleManager就可以;而如果你的业务逻辑不需要对数据进行排序,那么建议参考后面的几个参数调优,通过bypass机制或优化的HashShuffleManager来避免排序操作,同时提供较好的磁盘读写性能。这里要注意的是,tungsten-sort要慎用,因为之前发现了一些相应的bug。


    7、Shuffle优化配置 - spark.shuffle.sort.bypassMergeThreshold

    默认值:200

    参数说明:当ShuffleManager为SortShuffleManager时,如果shuffle read task的数量小于这个阈值(默认是200),则shuffle write过程中不会进行排序操作,而是直接按照未经优化的HashShuffleManager的方式去写数据,但是最后会将每个task产生的所有临时磁盘文件都合并成一个文件,并会创建单独的索引文件。

    调优建议:当你使用SortShuffleManager时,如果的确不需要排序操作,那么建议将这个参数调大一些,大于shuffle read task的数量。那么此时就会自动启用bypass机制,map-side就不会进行排序了,减少了排序的性能开销。但是这种方式下,依然会产生大量的磁盘文件,因此shuffle write性能有待提高。


    8、Shuffle优化配置 - spark.shuffle.consolidateFiles

    默认值:false

    参数说明:如果使用HashShuffleManager,该参数有效。如果设置为true,那么就会开启consolidate机制,会大幅度合并shuffle write的输出文件,对于shuffle read task数量特别多的情况下,这种方法可以极大地减少磁盘IO开销,提升性能。

    调优建议:如果的确不需要SortShuffleManager的排序机制,那么除了使用bypass机制,还可以尝试将spark.shffle.manager参数手动指定为hash,使用HashShuffleManager,同时开启consolidate机制。在实践中尝试过,发现其性能比开启了bypass机制的SortShuffleManager要高出10%~30%。


    总结:

    1、spark.shuffle.file.buffer:主要是设置的Shuffle过程中写文件的缓冲,默认32k,如果内存足够,可以适当调大,来减少写入磁盘的数量。

    2、spark.reducer.maxSizeInFight:主要是设置Shuffle过程中读文件的缓冲区,一次能够读取多少数据,如果内存足够,可以适当扩大,减少整个网络传输次数。

    3、spark.shuffle.io.maxRetries:主要是设置网络连接失败时,重试次数,适当调大能够增加稳定性。

    4、spark.shuffle.io.retryWait:主要设置每次重试之间的间隔时间,可以适当调大,增加程序稳定性。

    5、spark.shuffle.memoryFraction:Shuffle过程中的内存占用,如果程序中较多使用了Shuffle操作,那么可以适当调大该区域。

    6、spark.shuffle.manager:Hash和Sort方式,Sort是默认,Hash在reduce数量 比较少的时候,效率会很高。

    7、spark.shuffle.sort. bypassMergeThreshold:设置的是Sort方式中,启用Hash输出方式的临界值,如果你的程序数据不需要排序,而且reduce数量比较少,那推荐可以适当增大临界值。

    8、spark. shuffle.cosolidateFiles:如果你使用Hash shuffle方式,推荐打开该配置,实现更少的文件输出。

     参考链接:

    https://www.cnblogs.com/haozhengfei/p/5fc4a976a864f33587b094f36b72c7d3.html

    二、运行资源 

    2、资源运行中的集中情况

    (1)实践中跑的Spark job,有的特别慢,查看CPU利用率很低,可以尝试减少每个executor占用CPU core的数量,增加并行的executor数量,同时配合增加分片,整体上增加了CPU的利用率,加快数据处理速度。

    (2)发现某job很容易发生内存溢出,我们就增大分片数量,从而减少了每片数据的规模,同时还减少并行的executor数量,这样相同的内存资源分配给数量更少的executor,相当于增加了每个task的内存分配,这样运行速度可能慢了些,但是总比OOM强。

    (3)数据量特别少,有大量的小文件生成,就减少文件分片,没必要创建那么多task,这种情况,如果只是最原始的input比较小,一般都能被注意到;但是,如果是在运算过程中,比如应用某个reduceBy或者某个filter以后,数据大量减少,这种低效情况就很少被留意到。


    3、运行资源优化配置

    一个CPU core同一时间只能执行一个线程。而每个Executor进程上分配到的多个task,都是以每个task一条线程的方式,多线程并发运行的

    一个应用提交的时候设置多大的内存?设置多少Core?设置几个Executor?

    ./bin/spark-submit

    --master yarn-cluster

    --num-executors 100

    --executor-memory 6G

    --executor-cores 4

    --driver-memory 1G

    --conf spark.default.parallelism=1000

    --conf spark.storage.memoryFraction=0.5

    --conf spark.shuffle.memoryFraction=0.3

     

    3.1 运行资源优化配置 - num-executors

    参数说明:该参数用于设置Spark作业总共要用多少个Executor进程来执行。Driver在向YARN集群管理器申请资源时,YARN集群管理器会尽可能按照你的设置来在集群的各个工作节点上,启动相应数量的Executor进程。这个参数非常之重要,如果不设置的话,默认只会给你启动少量的Executor进程,此时你的Spark作业的运行速度是非常慢的。

    参数调优建议:每个Spark作业的运行一般设置50~100个左右的Executor进程比较合适,设置太少或太多的Executor进程都不好。设置的太少,无法充分利用集群资源;设置的太多的话,大部分队列可能无法给予充分的资源。


    3.2 运行资源优化配置 - executor-memory

    参数说明:该参数用于设置每个Executor进程的内存。Executor内存的大小,很多时候直接决定了Spark作业的性能,而且跟常见的JVM OOM异常,也有直接的关联。

    参数调优建议:每个Executor进程的内存设置4G~8G较为合适。但是这只是一个参考值,具体的设置还是得根据不同部门的资源队列来定。可以看看自己团队的资源队列的最大内存限制是多少,num-executors * executor-memory,是不能超过队列的最大内存量的。此外,如果你是跟团队里其他人共享这个资源队列,那么申请的内存量最好不要超过资源队列最大总内存的1/3~1/2,避免你自己的Spark作业占用了队列所有的资源,导致别的同事的作业无法运行。


    3.3 运行资源优化配置 - executor-cores

    参数说明:该参数用于设置每个Executor进程的CPU core数量。这个参数决定了每个Executor进程并行执行task线程的能力。因为每个CPU core同一时间只能执行一个task线程,因此每个Executor进程的CPU core数量越多,越能够快速地执行完分配给自己的所有task线程。

    参数调优建议:Executor的CPU core数量设置为2~4个较为合适。同样得根据不同部门的资源队列来定,可以看看自己的资源队列的最大CPU core限制是多少,再依据设置的Executor数量,来决定每个Executor进程可以分配到几个CPU core。同样建议,如果是跟他人共享这个队列,那么num-executors * executor-cores不要超过队列总CPU core的1/3~1/2左右比较合适,也是避免影响其他同事的作业运行。


    3.4 运行资源优化配置 - driver-memory

    参数说明:该参数用于设置Driver进程的内存。

    参数调优建议:Driver的内存通常来说不设置,或者设置1G左右应该就够了。唯一需要注意的一点是,如果需要使用collect算子将RDD的数据全部拉取到Driver上进行处理(或者是用map side join操作),那么必须确保Driver的内存足够大,否则会出现OOM内存溢出的问题。


    3.5 运行资源优化配置 - spark.default.parallelism

    参数说明:该参数用于设置每个stage的默认task数量,也可以认为是分区数。这个参数极为重要,如果不设置可能会直接影响你的Spark作业性能。

    参数调优建议:Spark作业的默认task数量为500~1000个较为合适。很多人常犯的一个错误就是不去设置这个参数,那么此时就会导致Spark自己根据底层HDFS的block数量来设置task的数量,默认是一个HDFS block对应一个task。通常来说,Spark默认设置的数量是偏少的(比如就几十个task),如果task数量偏少的话,就会导致你前面设置好的Executor的参数都前功尽弃。试想一下,无论你的Executor进程有多少个,内存和CPU有多大,但是task只有1个或者10个,那么90%的Executor进程可能根本就没有task执行,也就是白白浪费了资源!因此Spark官网建议的设置原则是,设置该参数为num-executors * executor-cores的2~3倍较为合适,比如Executor的总CPU core数量为300个,那么设置1000个task是可以的,此时可以充分地利用Spark集群的资源。


    3.6 运行资源优化配置 - spark.storage.memoryFraction

    参数说明:该参数用于设置RDD持久化数据在Executor内存中能占的比例,默认是0.6。也就是说,默认Executor 60%的内存,可以用来保存持久化的RDD数据。根据你选择的不同的持久化策略,如果内存不够时,可能数据就不会持久化,或者数据会写入磁盘。

    参数调优建议:如果Spark作业中,有较多的RDD持久化操作,该参数的值可以适当提高一些,保证持久化的数据能够容纳在内存中。避免内存不够缓存所有的数据,导致数据只能写入磁盘中,降低了性能。但是如果Spark作业中的shuffle类操作比较多,而持久化操作比较少,那么这个参数的值适当降低一些比较合适。此外,如果发现作业由于频繁的gc导致运行缓慢(通过spark web ui可以观察到作业的gc耗时),意味着task执行用户代码的内存不够用,那么同样建议调低这个参数的值。


    3.7 运行资源优化配置 - spark.shuffle.memoryFraction

    参数说明:该参数用于设置shuffle过程中一个task拉取到上个stage的task的输出后,进行聚合操作时能够使用的Executor内存的比例,默认是0.2。也就是说,Executor默认只有20%的内存用来进行该操作。shuffle操作在进行聚合时,如果发现使用的内存超出了这个20%的限制,那么多余的数据就会溢写到磁盘文件中去,此时就会极大地降低性能。

    参数调优建议:如果Spark作业中的RDD持久化操作较少,shuffle操作较多时,建议降低持久化操作的内存占比,提高shuffle操作的内存占比比例,避免shuffle过程中数据过多时内存不够用,必须溢写到磁盘上,降低了性能。此外,如果发现作业由于频繁的gc导致运行缓慢,意味着task执行用户代码的内存不够用,那么同样建议调低这个参数的值。


    总结:

    1、num-executors:应用运行时executor的数量,推荐50-100左右比较合适

    2、executor-memory:应用运行时executor的内存,推荐4-8G比较合适

    3、executor-cores:应用运行时executor的CPU核数,推荐2-4个比较合适

    4、driver-memory:应用运行时driver的内存量,主要考虑如果使用map side join或者一些类似于collect的操作,那么要相应调大内存量

    5、spark.default.parallelism:每个stage默认的task数量,推荐参数为num-executors * executor-cores的2~3倍较为合适

    6、spark.storage.memoryFraction:每一个executor中用于RDD缓存的内存比例,如果程序中有大量的数据缓存,可以考虑调大整个的比例,默认为60%

    7、spark.shuffle.memoryFraction:每一个executor中用于Shuffle操作的内存比例,默认是20%,如果程序中有大量的Shuffle类算子,那么可以考虑其它的比例。

  • 相关阅读:
    ThreadLocal的原理和使用场景
    sleep wait yield join方法的区别
    GC如何判断对象可以被回收
    双亲委派
    ConcurrentHashMap原理,jdk7和jdk8版本
    hashCode和equals
    接口和抽象类区别
    为什么局部内部类和匿名内部类只能访问局部final变量
    【GAN】基础GAN代码解析
    TF相关codna常用命令整理
  • 原文地址:https://www.cnblogs.com/hejunhong/p/12562953.html
Copyright © 2011-2022 走看看