spark（二）优化思路

时间：2017-09-01 19:43:51 阅读：197 评论：0 收藏：0 [点我收藏+]

标签：cloudera 压缩访问 exe 应用 manage 资料场景 str

优化思路

内存优化

内存优化大概分为三个方向

1.所有对象的总内存（包括数据和java对象）

2.访问这些对象的开销

3.垃圾回收的开销

其中Java的原生对象往往都能被很快的访问，但是会多占据2-5倍或更多的内存，有下面4点原因

·每个单独的java对象都有一个对象头（16字节），其中包括指向对象的指针（栈->堆），如果该对象只有几个属性，那么对象头可能比实际数据占用的空间都大（严重浪费资源）

·java每个string都包含了40字节的额外开销（因为底层其实是存储在数组，需要记录数组的指针，长度等信息），每个字符包含2字节（UTF-16编码）。例如一个10字符的string，实际占用内存空间60字节

·常见的集合类，例如linkedlist，hashmap都有用到链表，其中的对象头，元素指针都会占据额外的空间

·基础类型的包装，例如Integer

明确内存消耗

一般情况，可以把数据转成rdd，然后通过spark自带的UI的Storage页面来观察rdd占用内存大小。

其它特殊的对象，可以用spark自带的工具类SizeEstimator来评估内存大小，包括对广播数据的内存占用评估

优化数据结构

避免使用需要额外开销的java原生的数据结构，比如链表，hashmap，包装类。下面是常见的方法

·尽量使用数组结构和基础类型，

·在嵌套的数据结构中，尽量避免小对象和指针

·考虑用数字或者枚举来代替string作为key

·如果内存少于32GB，可以优化JVM -XX:+UseCompressedOops ，OOP = “ordinary object pointer” 普通对象指针。可以让指针由8->4字节，压缩的一般是对象的相关指针（不是用来压缩数据的）。

一般建议场景是在分配给JVM的内存大小在[4G,32G]，如果小于4G，那么JVM会使用低虚拟地址空间（low virutal address space，64位下模拟32位），这样就不需要做压解压动作了。而对于大于32G，将采用默认的随机地址分配特性，进行压解压。

数据序列化

选择合适的序列化协议，一般而言用Kryo，比java原生序列化快很多

数据存储

RDD persistence API

通过把数据序列化至内存，或者磁盘，或者其他策略

GC优化

所有给spark的内存资源，有一部分是用于cache RDD的，剩下的用于jvm的堆和栈等使用。

默认的比例是cache RDD占总内存的60%，可以通过spark.storage.memoryFraction来更改。

一般情况下，官方文档建议这个比值不要超过JVM Old Gen区域的比值。这也很容易理解，因为RDD Cache数据通常都是长期驻留内存的，理论上也就是说最终会被转移到Old Gen区域（如果该RDD还没有被删除的话），如果这部分数据允许的尺寸太大，势必把Old Gen区域占满，造成频繁的FULL GC，这种情况就可以调小该值。

·确认资源是否给足driver的cpu和memory，executor的cpu和memory

·当出现过多的full GC时候，可以减小RDD cache的内存空间

·当出现过多的minor GC时候，可以增加JVM中Eden区的大小，通过4/3的比例增加

·其它常规JVM优化方法，线程栈的内存大小，永久代的堆内存大小等

分区数优化

分区数多就是task多，整个任务的并发度就高，但也不是越多越好，假设你有100条数据，有50个分区，平均一个分区就处理两条数据，这样就造成了严重的浪费，更多的时间浪费在分区间的shuffle，和driver的聚合。

下面是几个优化建议

·每个cpu core上跑2~3个tasks

·当task上的数据大于20KB的时候，可以考虑

·在当前的分区数的1.5倍来进行调优

关于分区

除了显示的声明rdd或者dataframe的分区数外，还有两种控制分区数的配置，

1.spark.sql.shuffle.partitions

针对dataframe和一些sql操作的分区数

默认的分区数为父RDD的最大分区数

2.spark.default.parallelism

针对rdd的默认分区数

一般分区数取决于executor的core数量，因为partition越多task越多，而task是spark的最小处理单元。executor的core数量不够，task再多也只能排队，反而慢了。

注：默认的shuffle后分区数为200

共享变量

广播

当遇到全局性的数据需要使用时，可以采用广播的方式

广播变量的优势：是因为不是每个task一份变量副本，而是变成每个节点的executor才一份副本。这样的话，就可以让变量产生的副本大大减少。

广播变量，初始的时候，就在Drvier上有一份副本。task在运行的时候，想要使用广播变量中的数据，此时首先会在自己本地的Executor对应的BlockManager中，尝试获取变量副本；如果本地没有，BlockManager，也许会从远程的Driver上面去获取变量副本；也有可能从距离比较近的其他节点的Executor的BlockManager上去获取，并保存在本地的BlockManager中；BlockManager负责管理某个Executor对应的内存和磁盘上的数据，此后这个executor上的task，都会直接使用本地的BlockManager中的副本。

例如，50个executor，1000个task。一个map，10M。默认情况下，1000个task，1000份副本。10G的数据，网络传输，在集群中，耗费10G的内存资源。如果使用了广播变量。50个execurtor，50个副本。500M的数据，网络传输，而且不一定都是从Driver传输到每个节点，还可能是就近从最近的节点的executor的bockmanager上拉取变量副本，网络传输速度大大增加；500M，大大降低了内存消耗。

累加器

全局的累加器，可以用于统计全局性的数据

数据本地化

数据本地化是一个影响spark jobs性能的主要方面。其实运行分为两块，一块数据，一块代码，最好的情况就是数据不动（数据量太大），代码会部署在各个executor上。

可以通过调节spark.locality相关配置来决定任务的运行选择。

接口优化

1.reduceBy和groupBy

同理，reduceByKey,aggregateByKey,groupByKey等

优先使用reduceBy

reduceBy会优先合并本地的rdd，这样就大大的减少了shuffle的数据量

2.coalesce和repartition

看源码可知repartition是采用shuffle的coalesce。从性能上来讲，coalesce是本地合并，也就是同一个executor合并，这样可以减少网络传输带来的性能损失，并且是窄依赖，数据恢复也方便。而reparation直接采用shuffle的方式合并。优先使用coalesce。但是在大量分区需要合并的时候，要考虑一下策略。比如，现在一共有1000个分区，需要合并成10个分区。

如果直接采用coalesce(10)，可能导致合并的速度并不快（原因未知），而采用reparation(10)并发度会多很多。最终性能还是repartition好一点。

应用场景，设原rdd分区大小为M，现rdd分区大小为N
M>N，并且差10倍以内，考虑用coalesce
M>N，并且差10倍以上，考虑用repartition
M<N，考虑用repartition

注：也可以采用混合使用，先coalesce，把分区数降下来，然后采用repartition，当然这就需要实际测试，观察哪种性能更加