小白学习Spark系列六:Spark调参优化

时间:2024-10-01 14:37:20

  前几节介绍了下常用的函数和常踩的坑以及如何打包程序,现在来说下如何调参优化。当我们开发完一个项目,测试完成后,就要提交到服务器上运行,但运行不稳定,老是抛出如下异常,这就很纳闷了呀,明明测试上没问题,咋一到线上就出bug了呢!别急,我们来看下这bug到底怎么回事~

一、错误分析

  1、参数设置及异常信息

小白学习Spark系列六:Spark调参优化

18/10/08 16:23:51 WARN TransportChannelHandler: Exception in connection from /10.200.2.95:40888
java.io.IOException: Connection reset by peer
at sun.nio.ch.FileDispatcherImpl.read0(Native Method)
at sun.nio.ch.SocketDispatcher.read(SocketDispatcher.java:39)
at sun.nio.ch.IOUtil.readIntoNativeBuffer(IOUtil.java:223)
at sun.nio.ch.IOUtil.read(IOUtil.java:192)
at sun.nio.ch.SocketChannelImpl.read(SocketChannelImpl.java:380)
at io.netty.buffer.PooledUnsafeDirectByteBuf.setBytes(PooledUnsafeDirectByteBuf.java:221)
at io.netty.buffer.AbstractByteBuf.writeBytes(AbstractByteBuf.java:899)
at io.netty.channel.socket.nio.NioSocketChannel.doReadBytes(NioSocketChannel.java:275)
at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:119)
at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:652)
at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:575)
at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:489)
at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:451)
at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:140)
at io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:144)
at java.lang.Thread.run(Thread.java:745)
18/10/08 16:23:51 ERROR TransportResponseHandler: Still have 1 requests outstanding when connection from /10.200.2.95:40888 is closed

  2、分析原因

  运行的程序其实逻辑上比较简单,只是从hive表里读取的数据量很大,差不多60+G,并且需要将某些hive表读取到dirver节点上,用来获取每个executor上某些数据的映射值,所以driver设定的资源较大。运行时抛出的异常信息,从网上查了下原因大致是服务器的并发连接数超过了其承载量,服务器会将其中一些连接Down掉,这也就是说在运行spark程序时,过多的申请资源并发执行。那应该怎样去合理设置参数才能最大化提升并发的性能呢?这些参数又分别代表什么?

二、常见参数及含义

常用参数

含义及建议

spark.executor.memory

(--executor-memory)

含义:每个执行器进程分配的内存

建议:该设置需要和下面的executor-num数一起考虑,一般设置为4G~8G如果和其他人共用队列,为防止独占资源建议memory*num <= 资源队列最大总内存*(1/2~1/3)

spark.executor.num

(--executor-num)

含义:设置用来执行spark作业的执行器进程的个数

建议:一般设置为50~100个,设置太少会导致无法充分利用资源,太多又导致大部分任务分配不到充足的资源

spark.executor.cores

(--executor-cores)

含义:每个执行器进程的cpu core数量,决定了执行器进程并发执行task线程的个数,一个core同一时间只能执行一个task线程。如果core数量越多,完成自己task越快。

建议:一般设置2~4个,需要和executor-num一起考虑,num*core<=队列总core*(1/2~1/3)

spark.driver.memory

(--driver-memory)

含义:设置驱动进程的内存

建议:driver一般不设置,默认的就可以,不过如果你在程序中需要使用collect算子拉取rdd到驱动节点上,那就得设置相应的内存大小(大于几十k建议使用广播变量)

spark.default.parallelism

含义:设置每个stage的默认任务数量

建议:官方建议设置为 executor-num*executor-cores的2~3倍

spark.storage.memoryFraction

含义:默认Executor 60%的内存,可以用来保存持久化的RDD数据

建议:spark作业中,如果有较多rdd需要持久化,该参数可适当提高一些,保证持久化的数据存储在内存中,避免被写入磁盘影响运行速度。如果shuffle类操作较多,可调低该参数。并且如果由于频繁的垃圾回收导致运行缓慢,证明执行task的内存不够用,建议调低该参数。

spark.shuffle.memoryFraction

含义:设置shuffle过程中一个task拉取到上个stage的task的输出后,进行聚合操作时能够使用的Executor内存的比例

建议:shuffle操作在进行聚合时,如果发现使用的内存超出了这个20%的限制,那么多余的数据就会溢写到磁盘文件中去,此时就会极大地降低性能。结合上一个参数调整。

spark.speculation

含义:设为true时开启task预测执行机制。当出现较慢的任务时,这种机制会在另外的节点尝试执行该任务的一个副本。

建议:true,打开此选项会减少大规模集群中个别较慢的任务带来的影响。

spark.storage.blockManagerTimeoutIntervalMs

含义:内部通过超时机制追踪执行器进程是否存活的阈值。

建议:对于会引发长时间垃圾回收(GC)暂停的作业,需要把这个值调到100秒(100000)以防止失败。

spark.sql.shuffle.partitions

含义:配置join或者聚合操作shuffle数据时分区的数量,默认200

建议:同spark.default.parallelism

三、实践

  通过适当调整以上讲到的几个参数,降低spark.default.parallelism的同时又设置了spark.sql.shuffle.partitions、spark.speculation、spark.storage.blockManagerTimeoutIntervalMs三个参数。由于项目中频繁的读取hive表数据,并进行连接操作,所以在shuffle阶段增大了partitions。对于woker倾斜,设置spark.speculation=true,把预测不乐观的节点去掉来保证程序可稳定运行,通过这几个参数的调整这样并大大减少了运行时间。

小白学习Spark系列六:Spark调参优化