spark使用性能优化记录

Posted lsbigdata

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了spark使用性能优化记录相关的知识,希望对你有一定的参考价值。

性能调优:

总则:加资源加并行度 简单直接,调节最优的资源配置 RDD架构和持久化

  当可分配的资源无法达到更多的时候在考虑性能调优

   重剑无锋 花拳绣腿

  1.分配资源 并行度 RDD架构和缓存

  2.shuffle调优

  3.spark算子调优

  4.JVM调优 、 广播大变量

    分配哪些资源:executor(task--worker任务数)  cpu per  executor(每个作业的cpu核心数)、memory (可以使用的内存)、driver memory(影响不大)

spark submit shell脚本里面调整对应的参数

yarn资源队列

standalone  熟悉硬件配置

 原则:你能使用多大资源就调节多大

可有并行执行的task ,executor * cpu cor,减少磁盘IO 提高性能

如果有RDD.cache(),分配更多的内存,不用写入磁盘 ,如果内存较小,jvm 垃圾回收也会成为较大的问题,shuffle阶段reduce端拉取数据聚合,也是消耗内存,溢出会写入磁盘。

Job: 多个stage ,stage0结束到reduceBykey的时候会在stage1阶段的reducetask建立一份文件(可以有多个key但是相同key的一定在同一个文件里) stage0 端拉取自己的数据

并行度:  各个stage阶段的task数量官方推荐 task数量设置成cpu核心数的 2-3倍  这样CPU core 不至于那样的空闲 提升性能conf.set("spark.default.parallelism","500")

RDD持久化后,内存、磁盘 各有一部分数据当内存不够支撑 可以将RDD序列化成字节数组,就一个对象大大减少内存空间占用,在使用需要反序列化 还是OOM 存入磁盘,内存很大时,可以双副本机制两份持久化数据

纯内存 persist(StorageLevel.MEMORY_ONLY())   可以用.cache()代替

   persist(StorageLevel.MEMORY_ONLY_SER())

   persist(StorageLevel.MEMORY_AND_DISK())

   persist(StorageLevel.MEMORY_AND_DISK_SER())

   persist(StorageLevel.DISK_ONLY())

   persist(StorageLevel.MEMORY_ONLY_2()) 双副本机制

广播变量:   不用的话map副本消耗网络资源传输到各个task 每个都占用内存,堆区消耗增大,大变量广播出去,不是直接使用,开始driver上面一份 taskBlockManager负责管理的某个Executoor对应的内存磁盘上的数据上面拿 ,BLM没有的话会去driver或者另一个BLM上面拿不是每个task一个副本而是每个executor上一个随机抽取的时候可以用到广播变量broadcast

Kryo序列化

 使用:算子中用到的外部变量、持久化RDD进行序列化时、shuffle

 Kryo序列化 比java默认的序列化速度快 内存占用小 是java序列化机制的1/10

在sparkconf中设置属性 .set(spark.serializer,org.apache.spark.serializer.KryoSreializer)

注册你需要Kryo序列化的类 .registerKryoClass(new Class[]{xxx.class})

数据本地化等待时间

task没有被分配到它数据所在的节点(节点计算资源满了)

task等待默认3s 等不下去就会被分配到别的节点 或者task去用节点的另一个executor执行,taskexecutorBlockManager拉取所需的数据,这需要网络传输,通过TransferService 去远程其他节点找。最差可能需要跨机架去拉取数据

本地化级别 进程本地化 、节点本地化、机架本地化

可以观察日志 PROCESS_LOCAL ANY 都是什么级别的 调节等待时长 查看spark作业时长有没有缩短

调节方法:conf .set("spark.locality","10")

JVM调优

降低cache操作的内存占比

task执行算子函数时 有更多的内存可以使用 减少persist内存占比

JVM 内存划分:算子计算用 、persist用于持久化

executor 堆外内存与连接等待时长

task stage0 阶段,堆外内存不足,OOM,block manager 挂掉

task stage1 通过mapoutputtracker 得到需要的数据地址stage0 阶段找的时候 找不到此时任务会重新执行...反复几次task失败查看log shuffle output file not found 就是这个问题 task lost oom

设置 :提交作业的时候参数设置里面默认是300M 通常大任务的时候是不够的

-conf spark.yarn.executor.memoryOverhead=2048

task去别的BlockManager上拉取数据的那个节点正在进行JVM GC,此时连接超过默认60s 超时失败反复提交stage

 报错: errorfile id uuid not found file lost

 -conf spark.core.connection.ack.wait.time.out=300,这个可能解决偶尔的数据拉取失败情况

shuffle调优

shuffle原理

在某个action触发job的时候 DAGScheduler 会负责划分job为多个stage 依据算子 将操作前半部分以及之前所有的RDD transformation操作划分为一个stage shuffle的后半部分直到action为之的RDDtransformation操作划分为另外一个stage所以一个shuffle 会有两个stage ,stage0 通过内存缓冲区,缓冲区满溢后再spill溢写将key相同的写入磁盘的同一个文件,driver总共有多少个task 他就生成多少个文件,stage1 将属于他的key的那个文件拉取文件K,V对 到内存缓冲区 用HashMap数据格式 进行(kv)的汇聚。task用我们自定义的聚合函数进行累加得到的最终的值 就完成了shuffle

合并map端输出文件

.set("spark.shuffle.consolidateFiles","true")   默认是不开启的

并行执行的task创建输出文件 下一批并行执行的task 复用之前已有的输出文件  大大减少了map端的输出文件也减少了stage1阶段task任务的拉取文件数大大减少了磁盘的IO,实际生产环节 可能减少的时间是可以缩短一半的

调节map端内存缓冲与reduce端内存占比

map端内存缓冲默认32K 如果task处理数量较大 需要频繁的将内存溢写到磁盘

reduce端内存使用自己executorJVM 堆空间q 分配是默认 0.2 当拉取的数据过多  

频繁的spill操作 溢写到磁盘 只有聚合的时候还的多次读写存在磁盘的数据进行聚合

参数查看: standalone spark UI :4040 stage  executor task shuffle read  writer

yarn  界面 application  spark UI ...

实现设置: .set("spark.shuffle.file.buffer"."32") ,调节 64 、128 成倍加

.set("spark.shuffle.memoryFraction""0.2")调节0.1 的加

调节之后,减少了内存缓冲区溢写的次数 ,也减少了聚合读取磁盘文件的数量 

HashShuffleManagerSortShuffleManager

spark.shuffle.managerhashsorttungsten-sort

new SparkConf().set("spark.shuffle.manager", "hash") 普通的规则

new SparkConf().set("spark.shuffle.manager", "tungsten-sort")钨丝,官方说是优化了内存机制

 

以上是关于spark使用性能优化记录的主要内容,如果未能解决你的问题,请参考以下文章

Spark记录-Spark性能优化解决方案

Spark记录-Spark性能优化(开发资源数据shuffle)

性能优化-记录工作中遇到的点滴

性能优化-记录工作中遇到的点滴

spark性能优化

万字宝典 |《 Spark性能优化全书》推荐收藏!