一零四六、Spark性能调优_spark参数优化
| 阿里云国内75折 回扣 微信号：monov8 |
| 阿里云国际，腾讯云国际，低至75折。AWS 93折 免费开户实名账号 代冲值 优惠多多 微信号：monov8 飞机：@monov6 |
本文分为四个部分基本涵盖了所有Spark优化的点面试和实际工作中必备。
- 《Spark性能优化开发调优篇》
- 《Spark性能优化资源调优篇》
- 《Spark性能优化数据倾斜调优篇》
- 《Spark性能优化shuffle调优篇》
Spark性能优化开发调优篇
在大数据计算领域Spark已经成为了越来越流行、越来越受欢迎的计算平台之一。Spark的功能涵盖了大数据领域的离线批处理、SQL类处理、流式/实时计算、机器学习、图计算等各种不同类型的计算操作应用范围与前景非常广泛。
然而通过Spark开发出高性能的大数据计算作业并不是那么简单的。如果没有对Spark作业进行合理的调优Spark作业的执行速度可能会很慢这样就完全体现不出Spark作为一种快速大数据计算引擎的优势来。因此想要用好Spark就必须对其进行合理的性能优化。
Spark的性能调优实际上是由很多部分组成的不是调节几个参数就可以立竿见影提升作业性能的。我们需要根据不同的业务场景以及数据情况对Spark作业进行综合性的分析然后进行多个方面的调节和优化才能获得最佳性能。
笔者根据之前的Spark作业开发经验以及实践积累总结出了一套Spark作业的性能优化方案。整套方案主要分为开发调优、资源调优、数据倾斜调优、shuffle调优几个部分。开发调优和资源调优是所有Spark作业都需要注意和遵循的一些基本原则是高性能Spark作业的基础数据倾斜调优主要讲解了一套完整的用来解决Spark作业数据倾斜的解决方案shuffle调优面向的是对Spark的原理有较深层次掌握和研究的同学主要讲解了如何对Spark作业的shuffle运行过程以及细节进行调优。
本文作为Spark性能优化指南的基础篇主要讲解开发调优以及资源调优。
开发调优 Spark性能优化的第一步就是要在开发Spark作业的过程中注意和应用一些性能优化的基本原则。开发调优就是要让大家了解以下一些Spark基本开发原则包括RDD lineage设计、算子的合理使用、特殊操作的优化等。在开发过程中时时刻刻都应该注意以上原则并将这些原则根据具体的业务以及实际的应用场景灵活地运用到自己的Spark作业中。
原则一避免创建重复的RDD
通常来说我们在开发一个Spark作业时首先是基于某个数据源比如Hive表或HDFS文件创建一个初始的RDD接着对这个RDD执行某个算子操作然后得到下一个RDD以此类推循环往复直到计算出最终我们需要的结果。在这个过程中多个RDD会通过不同的算子操作比如map、reduce等串起来这个“RDD串”就是RDD lineage也就是“RDD的血缘关系链”。我们在开发过程中要注意对于同一份数据只应该创建一个RDD不能创建多个RDD来代表同一份数据。一些Spark初学者在刚开始开发Spark作业时或者是有经验的工程师在开发RDD lineage极其冗长的Spark作业时可能会忘了自己之前对于某一份数据已经创建过一个RDD了从而导致对于同一份数据创建了多个RDD。这就意味着我们的Spark作业会进行多次重复计算来创建多个代表相同数据的RDD进而增加了作业的性能开销。
一个简单的例子
需要对名为“hello.txt”的HDFS文件进行一次map操作再进行一次reduce操作
//也就是说需要对一份数据执行两次算子操作。
//错误的做法对于同一份数据执行多次算子操作时创建多个RDD。
//这里执行了两次textFile方法针对同一个HDFS文件创建了两个RDD出来
//然后分别对每个RDD都执行了一个算子操作。
//这种情况下Spark需要从HDFS上两次加载hello.txt文件的内容并创建两个单独的RDD
//第二次加载HDFS文件以及创建RDD的性能开销很明显是白白浪费掉的。
val rdd1 = sc.textFile("[hdfs://192.168.0.1:9000/hello.txt](hdfs://192.168.0.1:9000/hello.txt)")
rdd1.map(...)
val rdd2 = sc.textFile("[hdfs://192.168.0.1:9000/hello.txt](hdfs://192.168.0.1:9000/hello.txt)")
rdd2.reduce(...)
//正确的用法对于一份数据执行多次算子操作时只使用一个RDD。
//这种写法很明显比上一种写法要好多了因为我们对于同一份数据只创建了一个RDD
//然后对这一个RDD执行了多次算子操作。
//但是要注意到这里为止优化还没有结束由于rdd1被执行了两次算子操作第二次执行reduce操作的时候
//还会再次从源头处重新计算一次rdd1的数据因此还是会有重复计算的性能开销。
//要彻底解决这个问题必须结合“原则三对多次使用的RDD进行持久化”
//才能保证一个RDD被多次使用时只被计算一次。
val rdd1 = sc.textFile("[hdfs://192.168.0.1:9000/hello.txt](hdfs://192.168.0.1:9000/hello.txt)")
rdd1.map(...)
rdd1.reduce(...)
原则二尽可能复用同一个RDD
除了要避免在开发过程中对一份完全相同的数据创建多个RDD之外在对不同的数据执行算子操作时还要尽可能地复用一个RDD。比如说有一个RDD的数据格式是key-value类型的另一个是单value类型的这两个RDD的value数据是完全一样的。那么此时我们可以只使用key-value类型的那个RDD因为其中已经包含了另一个的数据。对于类似这种多个RDD的数据有重叠或者包含的情况我们应该尽量复用一个RDD这样可以尽可能地减少RDD的数量从而尽可能减少算子执行的次数。
一个简单的例子
// 错误的做法。
// 有一个格式的RDD即rdd1。
// 接着由于业务需要对rdd1执行了一个map操作创建了一个rdd2
//而rdd2中的数据仅仅是rdd1中的value值而已也就是说rdd2是rdd1的子集。
JavaPairRDD rdd1 = ...
JavaRDD rdd2 = rdd1.map(...)
// 分别对rdd1和rdd2执行了不同的算子操作。
rdd1.reduceByKey(...)
rdd2.map(...)
// 正确的做法。
// 上面这个case中其实rdd1和rdd2的区别无非就是数据格式不同而已
//rdd2的数据完全就是rdd1的子集而已却创建了两个rdd并对两个rdd都执行了一次算子操作。
// 此时会因为对rdd1执行map算子来创建rdd2而多执行一次算子操作进而增加性能开销。
// 其实在这种情况下完全可以复用同一个RDD。
// 我们可以使用rdd1既做reduceByKey操作也做map操作。
// 在进行第二个map操作时只使用每个数据的tuple._2也就是rdd1中的value值即可。
JavaPairRDD rdd1 = ...
rdd1.reduceByKey(...)
rdd1.map(tuple._2...)
// 第二种方式相较于第一种方式而言很明显减少了一次rdd2的计算开销。
// 但是到这里为止优化还没有结束对rdd1我们还是执行了两次算子操作rdd1实际上还是会被计算两次。
// 因此还需要配合“原则三对多次使用的RDD进行持久化”进行使用
//才能保证一个RDD被多次使用时只被计算一次
原则三对多次使用的RDD进行持久化
当你在Spark代码中多次对一个RDD做了算子操作后恭喜你已经实现Spark作业第一步的优化了也就是尽可能复用RDD。此时就该在这个基础之上进行第二步优化了也就是要保证对一个RDD执行多次算子操作时这个RDD本身仅仅被计算一次。
Spark中对于一个RDD执行多次算子的默认原理是这样的每次你对一个RDD执行一个算子操作时都会重新从源头处计算一遍计算出那个RDD来然后再对这个RDD执行你的算子操作。这种方式的性能是很差的。
因此对于这种情况我们的建议是对多次使用的RDD进行持久化。此时Spark就会根据你的持久化策略将RDD中的数据保存到内存或者磁盘中。以后每次对这个RDD进行算子操作时都会直接从内存或磁盘中提取持久化的RDD数据然后执行算子而不会从源头处重新计算一遍这个RDD再执行算子操作。
对多次使用的RDD进行持久化的代码示例
/ 如果要对一个RDD进行持久化只要对这个RDD调用cache()和persist()即可。
// 正确的做法。
// cache()方法表示使用非序列化的方式将RDD中的数据全部尝试持久化到内存中。
// 此时再对rdd1执行两次算子操作时只有在第一次执行map算子时才会将这个rdd1从源头处计算一次。
// 第二次执行reduce算子时就会直接从内存中提取数据进行计算不会重复计算一个rdd。
val rdd1 = sc.textFile("[hdfs://192.168.0.1:9000/hello.txt](hdfs://192.168.0.1:9000/hello.txt)").cache()
rdd1.map(...)
rdd1.reduce(...)
// persist()方法表示手动选择持久化级别并使用指定的方式进行持久化。
// 比如说StorageLevel.MEMORY_AND_DISK_SER表示内存充足时优先持久化到内存中
//内存不充足时持久化到磁盘文件中。
// 而且其中的_SER后缀表示使用序列化的方式来保存RDD数据此时RDD中的每个partition
//都会序列化成一个大的字节数组然后再持久化到内存或磁盘中。
// 序列化的方式可以减少持久化的数据对内存/磁盘的占用量进而避免内存被持久化数据占用过多
//从而发生频繁GC。
val rdd1 = sc.textFile("[hdfs://192.168.0.1:9000/hello.txt](hdfs://192.168.0.1:9000/hello.txt)")
.persist(StorageLevel.MEMORY_AND_DISK_SER)
rdd1.map(...)
rdd1.reduce(...)
对于persist()方法而言我们可以根据不同的业务场景选择不同的持久化级别。
Spark的持久化级别
如何选择一种最合适的持久化策略
1、默认情况下性能最高的当然是MEMORY_ONLY但前提是你的内存必须足够足够大可以绰绰有余地存放下整个RDD的所有数据。因为不进行序列化与反序列化操作就避免了这部分的性能开销对这个RDD的后续算子操作都是基于纯内存中的数据的操作不需要从磁盘文件中读取数据性能也很高而且不需要复制一份数据副本并远程传送到其他节点上。但是这里必须要注意的是在实际的生产环境中恐怕能够直接用这种策略的场景还是有限的如果RDD中数据比较多时比如几十亿直接用这种持久化级别会导致JVM的OOM内存溢出异常。
2、如果使用MEMORY_ONLY级别时发生了内存溢出那么建议尝试使用MEMORY_ONLY_SER级别。该级别会将RDD数据序列化后再保存在内存中此时每个partition仅仅是一个字节数组而已大大减少了对象数量并降低了内存占用。这种级别比MEMORY_ONLY多出来的性能开销主要就是序列化与反序列化的开销。但是后续算子可以基于纯内存进行操作因此性能总体还是比较高的。此外可能发生的问题同上如果RDD中的数据量过多的话还是可能会导致OOM内存溢出的异常。
3、如果纯内存的级别都无法使用那么建议使用MEMORY_AND_DISK_SER策略而不是MEMORY_AND_DISK策略。因为既然到了这一步就说明RDD的数据量很大内存无法完全放下。序列化后的数据比较少可以节省内存和磁盘的空间开销。同时该策略会优先尽量尝试将数据缓存在内存中内存缓存不下才会写入磁盘。
4、通常不建议使用DISK_ONLY和后缀为_2的级别因为完全基于磁盘文件进行数据的读写会导致性能急剧降低有时还不如重新计算一次所有RDD。后缀为_2的级别必须将所有数据都复制一份副本并发送到其他节点上数据复制以及网络传输会导致较大的性能开销除非是要求作业的高可用性否则不建议使用。
原则四尽量避免使用shuffle类算子
如果有可能的话要尽量避免使用shuffle类算子。因为Spark作业运行过程中最消耗性能的地方就是shuffle过程。shuffle过程简单来说就是将分布在集群中多个节点上的同一个key拉取到同一个节点上进行聚合或join等操作。比如reduceByKey、join等算子都会触发shuffle操作。
shuffle过程中各个节点上的相同key都会先写入本地磁盘文件中然后其他节点需要通过网络传输拉取各个节点上的磁盘文件中的相同key。而且相同key都拉取到同一个节点进行聚合操作时还有可能会因为一个节点上处理的key过多导致内存不够存放进而溢写到磁盘文件中。因此在shuffle过程中可能会发生大量的磁盘文件读写的IO操作以及数据的网络传输操作。磁盘IO和网络数据传输也是shuffle性能较差的主要原因。
因此在我们的开发过程中能避免则尽可能避免使用reduceByKey、join、distinct、repartition等会进行shuffle的算子尽量使用map类的非shuffle算子。这样的话没有shuffle操作或者仅有较少shuffle操作的Spark作业可以大大减少性能开销。
Broadcast与map进行join代码示例
// 传统的join操作会导致shuffle操作。
// 因为两个RDD中相同的key都需要通过网络拉取到一个节点上由一个task进行join操作。
val rdd3 = rdd1.join(rdd2)
// Broadcast+map的join操作不会导致shuffle操作。
// 使用Broadcast将一个数据量较小的RDD作为广播变量。
val rdd2Data = rdd2.collect()
val rdd2DataBroadcast = sc.broadcast(rdd2Data)
// 在rdd1.map算子中可以从rdd2DataBroadcast中获取rdd2的所有数据。
// 然后进行遍历如果发现rdd2中某条数据的key与rdd1的当前数据的key是相同的
//那么就判定可以进行join。
// 此时就可以根据自己需要的方式将rdd1当前数据与rdd2中可以连接的数据
//拼接在一起String或Tuple。
val rdd3 = rdd1.map(rdd2DataBroadcast...)
// 注意以上操作建议仅仅在rdd2的数据量比较少比如几百M或者一两G的情况下使用。
// 因为每个Executor的内存中都会驻留一份rdd2的全量数据。
- 本文来自168大数据 >> http://www.bi168.cn/thread-35907-1-1.html
原则五使用map-side预聚合的shuffle操作
如果因为业务需要一定要使用shuffle操作无法用map类的算子来替代那么尽量使用可以map-side预聚合的算子。
所谓的map-side预聚合说的是在每个节点本地对相同的key进行一次聚合操作类似于MapReduce中的本地combiner。map-side预聚合之后每个节点本地就只会有一条相同的key因为多条相同的key都被聚合起来了。其他节点在拉取所有节点上的相同key时就会大大减少需要拉取的数据数量从而也就减少了磁盘IO以及网络传输开销。通常来说在可能的情况下建议使用reduceByKey或者aggregateByKey算子来替代掉groupByKey算子。因为reduceByKey和aggregateByKey算子都会使用用户自定义的函数对每个节点本地的相同key进行预聚合。而groupByKey算子是不会进行预聚合的全量的数据会在集群的各个节点之间分发和传输性能相对来说比较差。
比如下图就是典型的例子分别基于reduceByKey和groupByKey进行单词计数。其中第一张图是groupByKey的原理图可以看到没有进行任何本地聚合时所有数据都会在集群节点之间传输第二张图是reduceByKey的原理图可以看到每个节点本地的相同key数据都进行了预聚合然后才传输到其他节点上进行全局聚合。
原则六使用高性能的算子
除了shuffle相关的算子有优化原则之外其他的算子也都有着相应的优化原则。
使用reduceByKey/aggregateByKey替代groupByKey
详情见“原则五使用map-side预聚合的shuffle操作”。
使用mapPartitions替代普通map
mapPartitions类的算子一次函数调用会处理一个partition所有的数据而不是一次函数调用处理一条性能相对来说会高一些。但是有的时候使用mapPartitions会出现OOM内存溢出的问题。因为单次函数调用就要处理掉一个partition所有的数据如果内存不够垃圾回收时是无法回收掉太多对象的很可能出现OOM异常。所以使用这类操作时要慎重
使用foreachPartitions替代foreach
原理类似于“使用mapPartitions替代map”也是一次函数调用处理一个partition的所有数据而不是一次函数调用处理一条数据。在实践中发现foreachPartitions类的算子对性能的提升还是很有帮助的。比如在foreach函数中将RDD中所有数据写MySQL那么如果是普通的foreach算子就会一条数据一条数据地写每次函数调用可能就会创建一个数据库连接此时就势必会频繁地创建和销毁数据库连接性能是非常低下但是如果用foreachPartitions算子一次性处理一个partition的数据那么对于每个partition只要创建一个数据库连接即可然后执行批量插入操作此时性能是比较高的。实践中发现对于1万条左右的数据量写MySQL性能可以提升30%以上。
使用filter之后进行coalesce操作
通常对一个RDD执行filter算子过滤掉RDD中较多数据后比如30%以上的数据建议使用coalesce算子手动减少RDD的partition数量将RDD中的数据压缩到更少的partition中去。因为filter之后RDD的每个partition中都会有很多数据被过滤掉此时如果照常进行后续的计算其实每个task处理的partition中的数据量并不是很多有一点资源浪费而且此时处理的task越多可能速度反而越慢。因此用coalesce减少partition数量将RDD中的数据压缩到更少的partition之后只要使用更少的task即可处理完所有的partition。在某些场景下对于性能的提升会有一定的帮助。
使用repartitionAndSortWithinPartitions替代repartition与sort类操作
repartitionAndSortWithinPartitions是Spark官网推荐的一个算子官方建议如果需要在repartition重分区之后还要进行排序建议直接使用repartitionAndSortWithinPartitions算子。因为该算子可以一边进行重分区的shuffle操作一边进行排序。shuffle与sort两个操作同时进行比先shuffle再sort来说性能可能是要高的。
原则七广播大变量
有时在开发过程中会遇到需要在算子函数中使用外部变量的场景尤其是大变量比如100M以上的大集合那么此时就应该使用Spark的广播Broadcast功能来提升性能。
在算子函数中使用到外部变量时默认情况下Spark会将该变量复制多个副本通过网络传输到task中此时每个task都有一个变量副本。如果变量本身比较大的话比如100M甚至1G那么大量的变量副本在网络中传输的性能开销以及在各个节点的Executor中占用过多内存导致的频繁GC都会极大地影响性能。
因此对于上述情况如果使用的外部变量比较大建议使用Spark的广播功能对该变量进行广播。广播后的变量会保证每个Executor的内存中只驻留一份变量副本而Executor中的task执行时共享该Executor中的那份变量副本。这样的话可以大大减少变量副本的数量从而减少网络传输的性能开销并减少对Executor内存的占用开销降低GC的频率。
广播大变量的代码示例
// 以下代码在算子函数中使用了外部的变量。
// 此时没有做任何特殊操作每个task都会有一份list1的副本。
val list1 = ...
rdd1.map(list1...)
// 以下代码将list1封装成了Broadcast类型的广播变量。
// 在算子函数中使用广播变量时首先会判断当前task所在Executor内存中是否有变量副本。
// 如果有则直接使用如果没有则从Driver或者其他Executor节点上远程拉取一份放到本地Executor内存中。
// 每个Executor内存中就只会驻留一份广播变量副本。
val list1 = ...
val list1Broadcast = sc.broadcast(list1)
rdd1.map(list1Broadcast...)
原则八使用Kryo优化序列化性能
在Spark中主要有三个地方涉及到了序列化
1、在算子函数中使用到外部变量时该变量会被序列化后进行网络传输见“原则七广播大变量”中的讲解。2、将自定义的类型作为RDD的泛型类型时比如JavaRDDStudent是自定义类型所有自定义类型对象都会进行序列化。因此这种情况下也要求自定义的类必须实现Serializable接口。3、使用可序列化的持久化策略时比如MEMORY_ONLY_SERSpark会将RDD中的每个partition都序列化成一个大的字节数组。
对于这三种出现序列化的地方我们都可以通过使用Kryo序列化类库来优化序列化和反序列化的性能。Spark默认使用的是Java的序列化机制也就是ObjectOutputStream/ObjectInputStream API来进行序列化和反序列化。但是Spark同时支持使用Kryo序列化库Kryo序列化类库的性能比Java序列化类库的性能要高很多。官方介绍Kryo序列化机制比Java序列化机制性能高10倍左右。Spark之所以默认没有使用Kryo作为序列化类库是因为Kryo要求最好要注册所有需要进行序列化的自定义类型因此对于开发者来说这种方式比较麻烦。
以下是使用Kryo的代码示例我们只要设置序列化类再注册要序列化的自定义类型即可比如算子函数中使用到的外部变量类型、作为RDD泛型类型的自定义类型等
// 创建SparkConf对象。
val conf = new SparkConf().setMaster(...).setAppName(...)
// 设置序列化器为KryoSerializer。
conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
// 注册要序列化的自定义类型。
conf.registerKryoClasses(Array(classOf[MyClass1], classOf[MyClass2]))
原则九优化数据结构
Java中有三种类型比较耗费内存
1、对象每个Java对象都有对象头、引用等额外的信息因此比较占用内存空间。2、字符串每个字符串内部都有一个字符数组以及长度等额外信息。3、集合类型比如HashMap、LinkedList等因为集合类型内部通常会使用一些内部类来封装集合元素比如Map.Entry。
因此Spark官方建议在Spark编码实现中特别是对于算子函数中的代码尽量不要使用上述三种数据结构尽量使用字符串替代对象使用原始类型比如Int、Long替代字符串使用数组替代集合类型这样尽可能地减少内存占用从而降低GC频率提升性能。
但是在笔者的编码实践中发现要做到该原则其实并不容易。因为我们同时要考虑到代码的可维护性如果一个代码中完全没有任何对象抽象全部是字符串拼接的方式那么对于后续的代码维护和修改无疑是一场巨大的灾难。同理如果所有操作都基于数组实现而不使用HashMap、LinkedList等集合类型那么对于我们的编码难度以及代码可维护性也是一个极大的挑战。因此笔者建议在可能以及合适的情况下使用占用内存较少的数据结构但是前提是要保证代码的可维护性。
Spark性能优化资源调优篇
在开发完Spark作业之后就该为作业配置合适的资源了。Spark的资源参数基本都可以在spark-submit命令中作为参数设置。很多Spark初学者通常不知道该设置哪些必要的参数以及如何设置这些参数最后就只能胡乱设置甚至压根儿不设置。资源参数设置的不合理可能会导致没有充分利用集群资源作业运行会极其缓慢或者设置的资源过大队列没有足够的资源来提供进而导致各种异常。总之无论是哪种情况都会导致Spark作业的运行效率低下甚至根本无法运行。因此我们必须对Spark作业的资源使用原理有一个清晰的认识并知道在Spark作业运行过程中有哪些资源参数是可以设置的以及如何设置合适的参数值。
一、 Spark作业基本运行原理
详细原理见上图。我们使用spark-submit提交一个Spark作业之后这个作业就会启动一个对应的Driver进程。根据你使用的部署模式deploy-mode不同Driver进程可能在本地启动也可能在集群中某个工作节点上启动。Driver进程本身会根据我们设置的参数占有一定数量的内存和CPU core。而Driver进程要做的第一件事情就是向集群管理器可以是Spark Standalone集群也可以是其他的资源管理集群美团•大众点评使用的是YARN作为资源管理集群申请运行Spark作业需要使用的资源这里的资源指的就是Executor进程。YARN集群管理器会根据我们为Spark作业设置的资源参数在各个工作节点上启动一定数量的Executor进程每个Executor进程都占有一定数量的内存和CPU core。
在申请到了作业执行所需的资源之后Driver进程就会开始调度和执行我们编写的作业代码了。Driver进程会将我们编写的Spark作业代码分拆为多个stage每个stage执行一部分代码片段并为每个stage创建一批task然后将这些task分配到各个Executor进程中执行。task是最小的计算单元负责执行一模一样的计算逻辑也就是我们自己编写的某个代码片段只是每个task处理的数据不同而已。一个stage的所有task都执行完毕之后会在各个节点本地的磁盘文件中写入计算中间结果然后Driver就会调度运行下一个stage。下一个stage的task的输入数据就是上一个stage输出的中间结果。如此循环往复直到将我们自己编写的代码逻辑全部执行完并且计算完所有的数据得到我们想要的结果为止。
Spark是根据shuffle类算子来进行stage的划分。如果我们的代码中执行了某个shuffle类算子比如reduceByKey、join等那么就会在该算子处划分出一个stage界限来。可以大致理解为shuffle算子执行之前的代码会被划分为一个stageshuffle算子执行以及之后的代码会被划分为下一个stage。因此一个stage刚开始执行的时候它的每个task可能都会从上一个stage的task所在的节点去通过网络传输拉取需要自己处理的所有key然后对拉取到的所有相同的key使用我们自己编写的算子函数执行聚合操作比如reduceByKey()算子接收的函数。这个过程就是shuffle。
当我们在代码中执行了cache/persist等持久化操作时根据我们选择的持久化级别的不同每个task计算出来的数据也会保存到Executor进程的内存或者所在节点的磁盘文件中。
因此Executor的内存主要分为三块第一块是让task执行我们自己编写的代码时使用默认是占Executor总内存的20%第二块是让task通过shuffle过程拉取了上一个stage的task的输出后进行聚合等操作时使用默认也是占Executor总内存的20%第三块是让RDD持久化时使用默认占Executor总内存的60%。
task的执行速度是跟每个Executor进程的CPU core数量有直接关系的。一个CPU core同一时间只能执行一个线程。而每个Executor进程上分配到的多个task都是以每个task一条线程的方式多线程并发运行的。如果CPU core数量比较充足而且分配到的task数量比较合理那么通常来说可以比较快速和高效地执行完这些task线程。
以上就是Spark作业的基本运行原理的说明大家可以结合上图来理解。理解作业基本原理是我们进行资源参数调优的基本前提。
二、 资源参数调优
了解完了Spark作业运行的基本原理之后对资源相关的参数就容易理解了。所谓的Spark资源参数调优其实主要就是对Spark运行过程中各个使用资源的地方通过调节各种参数来优化资源使用的效率从而提升Spark作业的执行性能。以下参数就是Spark中主要的资源参数每个参数都对应着作业运行原理中的某个部分我们同时也给出了一个调优的参考值。
num-executors
参数说明该参数用于设置Spark作业总共要用多少个Executor进程来执行。Driver在向YARN集群管理器申请资源时YARN集群管理器会尽可能按照你的设置来在集群的各个工作节点上启动相应数量的Executor进程。这个参数非常之重要如果不设置的话默认只会给你启动少量的Executor进程此时你的Spark作业的运行速度是非常慢的。
参数调优建议每个Spark作业的运行一般设置50~100个左右的Executor进程比较合适设置太少或太多的Executor进程都不好。设置的太少无法充分利用集群资源设置的太多的话大部分队列可能无法给予充分的资源。
executor-memory
参数说明该参数用于设置每个Executor进程的内存。Executor内存的大小很多时候直接决定了Spark作业的性能而且跟常见的JVM OOM异常也有直接的关联。
参数调优建议每个Executor进程的内存设置4G-8G较为合适。但是这只是一个参考值具体的设置还是得根据不同部门的资源队列来定。可以看看自己团队的资源队列的最大内存限制是多少num-executors乘以executor-memory就代表了你的Spark作业申请到的总内存量也就是所有Executor进程的内存总和这个量是不能超过队列的最大内存量的。此外如果你是跟团队里其他人共享这个资源队列那么申请的总内存量最好不要超过资源队列最大总内存的1/3~1/2避免你自己的Spark作业占用了队列所有的资源导致别的同学的作业无法运行。
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左右比较合适也是避免影响其他同学的作业运行。
driver-memory
参数说明该参数用于设置Driver进程的内存。
参数调优建议Driver的内存通常来说不设置或者设置1G左右应该就够了。唯一需要注意的一点是如果需要使用collect算子将RDD的数据全部拉取到Driver上进行处理那么必须确保Driver的内存足够大否则会出现OOM内存溢出的问题。
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集群的资源。
spark.storage.memoryFraction
参数说明该参数用于设置RDD持久化数据在Executor内存中能占的比例默认是0.6。也就是说默认Executor 60%的内存可以用来保存持久化的RDD数据。根据你选择的不同的持久化策略如果内存不够时可能数据就不会持久化或者数据会写入磁盘。
参数调优建议如果Spark作业中有较多的RDD持久化操作该参数的值可以适当提高一些保证持久化的数据能够容纳在内存中。避免内存不够缓存所有的数据导致数据只能写入磁盘中降低了性能。但是如果Spark作业中的shuffle类操作比较多而持久化操作比较少那么这个参数的值适当降低一些比较合适。此外如果发现作业由于频繁的gc导致运行缓慢通过spark web ui可以观察到作业的gc耗时意味着task执行用户代码的内存不够用那么同样建议调低这个参数的值。
spark.shuffle.memoryFraction
参数说明该参数用于设置shuffle过程中一个task拉取到上个stage的task的输出后进行聚合操作时能够使用的Executor内存的比例默认是0.2。也就是说Executor默认只有20%的内存用来进行该操作。shuffle操作在进行聚合时如果发现使用的内存超出了这个20%的限制那么多余的数据就会溢写到磁盘文件中去此时就会极大地降低性能。
参数调优建议如果Spark作业中的RDD持久化操作较少shuffle操作较多时建议降低持久化操作的内存占比提高shuffle操作的内存占比比例避免shuffle过程中数据过多时内存不够用必须溢写到磁盘上降低了性能。此外如果发现作业由于频繁的gc导致运行缓慢意味着task执行用户代码的内存不够用那么同样建议调低这个参数的值。资源参数的调优没有一个固定的值需要同学们根据自己的实际情况包括Spark作业中的shuffle操作数量、RDD持久化操作数量以及spark web ui中显示的作业gc情况同时参考本篇文章中给出的原理以及调优建议合理地设置上述参数。
三、资源参数参考示例
以下是一份spark-submit命令的示例大家可以参考一下并根据自己的实际情况进行调节
./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 \
Spark性能优化数据倾斜调优
一、前言
继《Spark性能优化开发调优篇》和《Spark性能优化资源调优篇》讲解了每个Spark开发人员都必须熟知的开发调优与资源调优之后本文作为《Spark性能优化指南》的高级篇将深入分析数据倾斜调优与shuffle调优以解决更加棘手的性能问题。
二、数据倾斜调优
2.1、调优概述 有的时候我们可能会遇到大数据计算中一个最棘手的问题——数据倾斜此时Spark作业的性能会比期望差很多。数据倾斜调优就是使用各种技术方案解决不同类型的数据倾斜问题以保证Spark作业的性能。
2.2、数据倾斜发生时的现象 1、绝大多数task执行得都非常快但个别task执行极慢。比如总共有1000个task997个task都在1分钟之内执行完了但是剩余两三个task却要一两个小时。这种情况很常见。原本能够正常执行的Spark作业某天突然报出OOM内存溢出异常观察异常栈是我们写的业务代码造成的。这种情况比较少见。
2.3、数据倾斜发生的原理 数据倾斜的原理很简单在进行shuffle的时候必须将各个节点上相同的key拉取到某个节点上的一个task来进行处理比如按照key进行聚合或join等操作。此时如果某个key对应的数据量特别大的话就会发生数据倾斜。比如大部分key对应10条数据但是个别key却对应了100万条数据那么大部分task可能就只会分配到10条数据然后1秒钟就运行完了但是个别task可能分配到了100万数据要运行一两个小时。因此整个Spark作业的运行进度是由运行时间最长的那个task决定的。
因此出现数据倾斜的时候Spark作业看起来会运行得非常缓慢甚至可能因为某个task处理的数据量过大导致内存溢出。
下图就是一个很清晰的例子hello这个key在三个节点上对应了总共7条数据这些数据都会被拉取到同一个task中进行处理而world和you这两个key分别才对应1条数据所以另外两个task只要分别处理1条数据即可。此时第一个task的运行时间可能是另外两个task的7倍而整个stage的运行速度也由运行最慢的那个task所决定。
2.4、如何定位导致数据倾斜的代码 数据倾斜只会发生在shuffle过程中。这里给大家罗列一些常用的并且可能会触发shuffle操作的算子distinct、groupByKey、reduceByKey、aggregateByKey、join、cogroup、repartition等。出现数据倾斜时可能就是你的代码中使用了这些算子中的某一个所导致的。
2.5、 某个task执行特别慢的情况 首先要看的就是数据倾斜发生在第几个stage中。
如果是用yarn-client模式提交那么本地是直接可以看到log的可以在log中找到当前运行到了第几个stage如果是用yarn-cluster模式提交则可以通过Spark Web UI来查看当前运行到了第几个stage。此外无论是使用yarn-client模式还是yarn-cluster模式我们都可以在Spark Web UI上深入看一下当前这个stage各个task分配的数据量从而进一步确定是不是task分配的数据不均匀导致了数据倾斜。
比如下图中倒数第三列显示了每个task的运行时间。明显可以看到有的task运行特别快只需要几秒钟就可以运行完而有的task运行特别慢需要几分钟才能运行完此时单从运行时间上看就已经能够确定发生数据倾斜了。此外倒数第一列显示了每个task处理的数据量明显可以看到运行时间特别短的task只需要处理几百KB的数据即可而运行时间特别长的task需要处理几千KB的数据处理的数据量差了10倍。此时更加能够确定是发生了数据倾斜。
知道数据倾斜发生在哪一个stage之后接着我们就需要根据stage划分原理推算出来发生倾斜的那个stage对应代码中的哪一部分这部分代码中肯定会有一个shuffle类算子。精准推算stage与代码的对应关系需要对Spark的源码有深入的理解这里我们可以介绍一个相对简单实用的推算方法只要看到Spark代码中出现了一个shuffle类算子或者是Spark SQL的SQL语句中出现了会导致shuffle的语句比如group by语句那么就可以判定以那个地方为界限划分出了前后两个stage。
这里我们就以Spark最基础的入门程序——单词计数来举例如何用最简单的方法大致推算出一个stage对应的代码。如下示例在整个代码中只有一个reduceByKey是会发生shuffle的算子因此就可以认为以这个算子为界限会划分出前后两个stage。
1、stage0主要是执行从textFile到map操作以及执行shuffle write操作。shuffle write操作我们可以简单理解为对pairs RDD中的数据进行分区操作每个task处理的数据中相同的key会写入同一个磁盘文件内。
2、stage1主要是执行从reduceByKey到collect操作stage1的各个task一开始运行就会首先执行shuffle read操作。执行shuffle read操作的task会从stage0的各个task所在节点拉取属于自己处理的那些key然后对同一个key进行全局性的聚合或join等操作在这里就是对key的value值进行累加。stage1在执行完reduceByKey算子之后就计算出了最终的wordCounts RDD然后会执行collect算子将所有数据拉取到Driver上供我们遍历和打印输出。
val conf = new SparkConf()
val sc = new SparkContext(conf)
val lines = sc.textFile("[hdfs://...](hdfs://...)")
val words = lines.flatMap(_.split(" "))
val pairs = words.map((_, 1))
val wordCounts = pairs.reduceByKey(_ + _)
wordCounts.collect().foreach(println(_))
通过对单词计数程序的分析希望能够让大家了解最基本的stage划分的原理以及stage划分后shuffle操作是如何在两个stage的边界处执行的。然后我们就知道如何快速定位出发生数据倾斜的stage对应代码的哪一个部分了。比如我们在Spark Web UI或者本地log中发现stage1的某几个task执行得特别慢判定stage1出现了数据倾斜那么就可以回到代码中定位出stage1主要包括了reduceByKey这个shuffle类算子此时基本就可以确定是由reduceByKey算子导致的数据倾斜问题。比如某个单词出现了100万次其他单词才出现10次那么stage1的某个task就要处理100万数据整个stage的速度就会被这个task拖慢。
2.6、某个task莫名其妙内存溢出的情况 这种情况下去定位出问题的代码就比较容易了。我们建议直接看yarn-client模式下本地log的异常栈或者是通过YARN查看yarn-cluster模式下的log中的异常栈。一般来说通过异常栈信息就可以定位到你的代码中哪一行发生了内存溢出。然后在那行代码附近找找一般也会有shuffle类算子此时很可能就是这个算子导致了数据倾斜。
但是大家要注意的是不能单纯靠偶然的内存溢出就判定发生了数据倾斜。因为自己编写的代码的bug以及偶然出现的数据异常也可能会导致内存溢出。因此还是要按照上面所讲的方法通过Spark Web UI查看报错的那个stage的各个task的运行时间以及分配的数据量才能确定是否是由于数据倾斜才导致了这次内存溢出。
2.7、查看导致数据倾斜的key的数据分布情况 知道了数据倾斜发生在哪里之后通常需要分析一下那个执行了shuffle操作并且导致了数据倾斜的RDD/Hive表查看一下其中key的分布情况。这主要是为之后选择哪一种技术方案提供依据。针对不同的key分布与不同的shuffle算子组合起来的各种情况可能需要选择不同的技术方案来解决。
此时根据你执行操作的情况不同可以有很多种查看key分布的方式
1、如果是Spark SQL中的group by、join语句导致的数据倾斜那么就查询一下SQL中使用的表的key分布情况。
2、如果是对Spark RDD执行shuffle算子导致的数据倾斜那么可以在Spark作业中加入查看key分布的代码比如RDD.countByKey()。然后对统计出来的各个key出现的次数collect/take到客户端打印一下就可以看到key的分布情况。
举例来说对于上面所说的单词计数程序如果确定了是stage1的reduceByKey算子导致了数据倾斜那么就应该看看进行reduceByKey操作的RDD中的key分布情况在这个例子中指的就是pairs RDD。如下示例我们可以先对pairs采样10%的样本数据然后使用countByKey算子统计出每个key出现的次数最后在客户端遍历和打印样本数据中各个key的出现次数。
val sampledPairs = pairs.sample(false, 0.1)
val sampledWordCounts = sampledPairs.countByKey()
sampledWordCounts.foreach(println(_))
三、数据倾斜的解决方案
解决方案一使用Hive ETL预处理数据
方案适用场景导致数据倾斜的是Hive表。如果该Hive表中的数据本身很不均匀比如某个key对应了100万数据其他key才对应了10条数据而且业务场景需要频繁使用Spark对Hive表执行某个分析操作那么比较适合使用这种技术方案。
方案实现思路此时可以评估一下是否可以通过Hive来进行数据预处理即通过Hive ETL预先对数据按照key进行聚合或者是预先和其他表进行join然后在Spark作业中针对的数据源就不是原来的Hive表了而是预处理后的Hive表。此时由于数据已经预先进行过聚合或join操作了那么在Spark作业中也就不需要使用原先的shuffle类算子执行这类操作了。
方案实现原理这种方案从根源上解决了数据倾斜因为彻底避免了在Spark中执行shuffle类算子那么肯定就不会有数据倾斜的问题了。但是这里也要提醒一下大家这种方式属于治标不治本。因为毕竟数据本身就存在分布不均匀的问题所以Hive ETL中进行group by或者join等shuffle操作时还是会出现数据倾斜导致Hive ETL的速度很慢。我们只是把数据倾斜的发生提前到了Hive ETL中避免Spark程序发生数据倾斜而已。
方案优点实现起来简单便捷效果还非常好完全规避掉了数据倾斜Spark作业的性能会大幅度提升。
方案缺点治标不治本Hive ETL中还是会发生数据倾斜。
方案实践经验在一些Java系统与Spark结合使用的项目中会出现Java代码频繁调用Spark作业的场景而且对Spark作业的执行性能要求很高就比较适合使用这种方案。将数据倾斜提前到上游的Hive ETL每天仅执行一次只有那一次是比较慢的而之后每次Java调用Spark作业时执行速度都会很快能够提供更好的用户体验。
项目实践经验在美团·点评的交互式用户行为分析系统中使用了这种方案该系统主要是允许用户通过Java Web系统提交数据分析统计任务后端通过Java提交Spark作业进行数据分析统计。要求Spark作业速度必须要快尽量在10分钟以内否则速度太慢用户体验会很差。所以我们将有些Spark作业的shuffle操作提前到了Hive ETL中从而让Spark直接使用预处理的Hive中间表尽可能地减少Spark的shuffle操作大幅度提升了性能将部分作业的性能提升了6倍以上。
解决方案二过滤少数导致倾斜的key
方案适用场景如果发现导致倾斜的key就少数几个而且对计算本身的影响并不大的话那么很适合使用这种方案。比如99%的key就对应10条数据但是只有一个key对应了100万数据从而导致了数据倾斜。
方案实现思路如果我们判断那少数几个数据量特别多的key对作业的执行和计算结果不是特别重要的话那么干脆就直接过滤掉那少数几个key。比如在Spark SQL中可以使用where子句过滤掉这些key或者在Spark Core中对RDD执行filter算子过滤掉这些key。如果需要每次作业执行时动态判定哪些key的数据量最多然后再进行过滤那么可以使用sample算子对RDD进行采样然后计算出每个key的数量取数据量最多的key过滤掉即可。
方案实现原理将导致数据倾斜的key给过滤掉之后这些key就不会参与计算了自然不可能产生数据倾斜。
方案优点实现简单而且效果也很好可以完全规避掉数据倾斜。
方案缺点适用场景不多大多数情况下导致倾斜的key还是很多的并不是只有少数几个。
方案实践经验在项目中我们也采用过这种方案解决数据倾斜。有一次发现某一天Spark作业在运行的时候突然OOM了追查之后发现是Hive表中的某一个key在那天数据异常导致数据量暴增。因此就采取每次执行前先进行采样计算出样本中数据量最大的几个key之后直接在程序中将那些key给过滤掉。
解决方案三提高shuffle操作的并行度
方案适用场景如果我们必须要对数据倾斜迎难而上那么建议优先使用这种方案因为这是处理数据倾斜最简单的一种方案。
方案实现思路在对RDD执行shuffle算子时给shuffle算子传入一个参数比如reduceByKey(1000)该参数就设置了这个shuffle算子执行时shuffle read task的数量。对于Spark SQL中的shuffle类语句比如group by、join等需要设置一个参数即spark.sql.shuffle.partitions该参数代表了shuffle read task的并行度该值默认是200对于很多场景来说都有点过小。
方案实现原理增加shuffle read task的数量可以让原本分配给一个task的多个key分配给多个task从而让每个task处理比原来更少的数据。举例来说如果原本有5个key每个key对应10条数据这5个key都是分配给一个task的那么这个task就要处理50条数据。而增加了shuffle read task以后每个task就分配到一个key即每个task就处理10条数据那么自然每个task的执行时间都会变短了。具体原理如下图所示。
方案优点实现起来比较简单可以有效缓解和减轻数据倾斜的影响。
方案缺点只是缓解了数据倾斜而已没有彻底根除问题根据实践经验来看其效果有限。
方案实践经验该方案通常无法彻底解决数据倾斜因为如果出现一些极端情况比如某个key对应的数据量有100万那么无论你的task数量增加到多少这个对应着100万数据的key肯定还是会分配到一个task中去处理因此注定还是会发生数据倾斜的。所以这种方案只能说是在发现数据倾斜时尝试使用的第一种手段尝试去用嘴简单的方法缓解数据倾斜而已或者是和其他方案结合起来使用。
解决方案四两阶段聚合局部聚合+全局聚合
方案适用场景对RDD执行reduceByKey等聚合类shuffle算子或者在Spark SQL中使用group by语句进行分组聚合时比较适用这种方案。
方案实现思路这个方案的核心实现思路就是进行两阶段聚合。第一次是局部聚合先给每个key都打上一个随机数比如10以内的随机数此时原先一样的key就变成不一样的了比如(hello, 1) (hello, 1) (hello, 1) (hello, 1)就会变成(1_hello, 1) (1_hello, 1) (2_hello, 1) (2_hello, 1)。接着对打上随机数后的数据执行reduceByKey等聚合操作进行局部聚合那么局部聚合结果就会变成了(1_hello, 2) (2_hello, 2)。然后将各个key的前缀给去掉就会变成(hello,2)(hello,2)再次进行全局聚合操作就可以得到最终结果了比如(hello, 4)。
方案实现原理将原本相同的key通过附加随机前缀的方式变成多个不同的key就可以让原本被一个task处理的数据分散到多个task上去做局部聚合进而解决单个task处理数据量过多的问题。接着去除掉随机前缀再次进行全局聚合就可以得到最终的结果。具体原理见下图。
方案优点对于聚合类的shuffle操作导致的数据倾斜效果是非常不错的。通常都可以解决掉数据倾斜或者至少是大幅度缓解数据倾斜将Spark作业的性能提升数倍以上。
方案缺点仅仅适用于聚合类的shuffle操作适用范围相对较窄。如果是join类的shuffle操作还得用其他的解决方案。
// 第一步给RDD中的每个key都打上一个随机前缀。
JavaPairRDD randomPrefixRdd = rdd.mapToPair(
new PairFunction, String, Long>() {
private static final long serialVersionUID = 1L;
@Override
public Tuple2 call(Tuple2 tuple)
throws Exception {
Random random = new Random();
int prefix = random.nextInt(10);
return new Tuple2(prefix + "_" + tuple._1, tuple._2);
}
});
// 第二步对打上随机前缀的key进行局部聚合。
JavaPairRDD localAggrRdd = randomPrefixRdd.reduceByKey(
new Function2() {
private static final long serialVersionUID = 1L;
@Override
public Long call(Long v1, Long v2) throws Exception {
return v1 + v2;
}
});
// 第三步去除RDD中每个key的随机前缀。
JavaPairRDD removedRandomPrefixRdd = localAggrRdd.mapToPair(
new PairFunction, Long, Long>() {
private static final long serialVersionUID = 1L;
@Override
public Tuple2 call(Tuple2 tuple)
throws Exception {
long originalKey = Long.valueOf(tuple._1.split("_")[1]);
return new Tuple2(originalKey, tuple._2);
}
});
// 第四步对去除了随机前缀的RDD进行全局聚合。
JavaPairRDD globalAggrRdd = removedRandomPrefixRdd.reduceByKey(
new Function2() {
private static final long serialVersionUID = 1L;
@Override
public Long call(Long v1, Long v2) throws Exception {
return v1 + v2;
}
});
解决方案五将reduce join转为map join
方案适用场景在对RDD使用join类操作或者是在Spark SQL中使用join语句时而且join操作中的一个RDD或表的数据量比较小比如几百M或者一两G比较适用此方案。
方案实现思路不使用join算子进行连接操作而使用Broadcast变量与map类算子实现join操作进而完全规避掉shuffle类的操作彻底避免数据倾斜的发生和出现。将较小RDD中的数据直接通过collect算子拉取到Driver端的内存中来然后对其创建一个Broadcast变量接着对另外一个RDD执行map类算子在算子函数内从Broadcast变量中获取较小RDD的全量数据与当前RDD的每一条数据按照连接key进行比对如果连接key相同的话那么就将两个RDD的数据用你需要的方式连接起来。
方案实现原理普通的join是会走shuffle过程的而一旦shuffle就相当于会将相同key的数据拉取到一个shuffle read task中再进行join此时就是reduce join。但是如果一个RDD是比较小的则可以采用广播小RDD全量数据+map算子来实现与join同样的效果也就是map join此时就不会发生shuffle操作也就不会发生数据倾斜。具体原理如下图所示。
方案优点对join操作导致的数据倾斜效果非常好因为根本就不会发生shuffle也就根本不会发生数据倾斜。
方案缺点适用场景较少因为这个方案只适用于一个大表和一个小表的情况。毕竟我们需要将小表进行广播此时会比较消耗内存资源driver和每个Executor内存中都会驻留一份小RDD的全量数据。如果我们广播出去的RDD数据比较大比如10G以上那么就可能发生内存溢出了。因此并不适合两个都是大表的情况。
// 首先将数据量比较小的RDD的数据collect到Driver中来。
List> rdd1Data = rdd1.collect()
// 然后使用Spark的广播功能将小RDD的数据转换成广播变量这样每个Executor就只有一份RDD的数据。
// 可以尽可能节省内存空间并且减少网络传输性能开销。
final Broadcast>> rdd1DataBroadcast = sc.broadcast(rdd1Data);
// 对另外一个RDD执行map类操作而不再是join类操作。
JavaPairRDD> joinedRdd = rdd2.mapToPair(
new PairFunction, String, Tuple2>() {
private static final long serialVersionUID = 1L;
@Override
public Tuple2> call(Tuple2 tuple)
throws Exception {
// 在算子函数中通过广播变量获取到本地Executor中的rdd1数据。
List> rdd1Data = rdd1DataBroadcast.value();
// 可以将rdd1的数据转换为一个Map便于后面进行join操作。
Map rdd1DataMap = new HashMap();
for(Tuple2 data : rdd1Data) {
rdd1DataMap.put(data._1, data._2);
}
// 获取当前RDD数据的key以及value。
String key = tuple._1;
String value = tuple._2;
// 从rdd1数据Map中根据key获取到可以join到的数据。
Row rdd1Value = rdd1DataMap.get(key);
return new Tuple2(key, new Tuple2(value, rdd1Value));
}
});
// 这里得提示一下。
// 上面的做法仅仅适用于rdd1中的key没有重复全部是唯一的场景。
// 如果rdd1中有多个相同的key那么就得用flatMap类的操作在进行join的时候不能用map而是得遍历rdd1所有数据进行join。
// rdd2中每条数据都可能会返回多条join后的数据。
- 本文来自168大数据 >> http://www.bi168.cn/thread-35907-1-1.html
解决方案六采样倾斜key并分拆join操作
方案适用场景两个RDD/Hive表进行join的时候如果数据量都比较大无法采用“解决方案五”那么此时可以看一下两个RDD/Hive表中的key分布情况。如果出现数据倾斜是因为其中某一个RDD/Hive表中的少数几个key的数据量过大而另一个RDD/Hive表中的所有key都分布比较均匀那么采用这个解决方案是比较合适的。
方案实现思路
1、对包含少数几个数据量过大的key的那个RDD通过sample算子采样出一份样本来然后统计一下每个key的数量计算出来数据量最大的是哪几个key。2、然后将这几个key对应的数据从原来的RDD中拆分出来形成一个单独的RDD并给每个key都打上n以内的随机数作为前缀而不会导致倾斜的大部分key形成另外一个RDD。3、接着将需要join的另一个RDD也过滤出来那几个倾斜key对应的数据并形成一个单独的RDD将每条数据膨胀成n条数据这n条数据都按顺序附加一个0~n的前缀不会导致倾斜的大部分key也形成另外一个RDD。4、再将附加了随机前缀的独立RDD与另一个膨胀n倍的独立RDD进行join此时就可以将原先相同的key打散成n份分散到多个task中去进行join了。5、而另外两个普通的RDD就照常join即可。6、最后将两次join的结果使用union算子合并起来即可就是最终的join结果。
方案实现原理对于join导致的数据倾斜如果只是某几个key导致了倾斜可以将少数几个key分拆成独立RDD并附加随机前缀打散成n份去进行join此时这几个key对应的数据就不会集中在少数几个task上而是分散到多个task进行join了。具体原理见下图。
方案优点对于join导致的数据倾斜如果只是某几个key导致了倾斜采用该方式可以用最有效的方式打散key进行join。而且只需要针对少数倾斜key对应的数据进行扩容n倍不需要对全量数据进行扩容。避免了占用过多内存。
方案缺点如果导致倾斜的key特别多的话比如成千上万个key都导致数据倾斜那么这种方式也不适合。
// 首先从包含了少数几个导致数据倾斜key的rdd1中采样10%的样本数据。
JavaPairRDD sampledRDD = rdd1.sample(false, 0.1);
// 对样本数据RDD统计出每个key的出现次数并按出现次数降序排序。
// 对降序排序后的数据取出top 1或者top 100的数据也就是key最多的前n个数据。
// 具体取出多少个数据量最多的key由大家自己决定我们这里就取1个作为示范。
JavaPairRDD mappedSampledRDD = sampledRDD.mapToPair(
new PairFunction, Long, Long>() {
private static final long serialVersionUID = 1L;
@Override
public Tuple2 call(Tuple2 tuple)
throws Exception {
return new Tuple2(tuple._1, 1L);
}
});
JavaPairRDD countedSampledRDD = mappedSampledRDD.reduceByKey(
new Function2() {
private static final long serialVersionUID = 1L;
@Override
public Long call(Long v1, Long v2) throws Exception {
return v1 + v2;
}
});
JavaPairRDD reversedSampledRDD = countedSampledRDD.mapToPair(
new PairFunction, Long, Long>() {
private static final long serialVersionUID = 1L;
@Override
public Tuple2 call(Tuple2 tuple)
throws Exception {
return new Tuple2(tuple._2, tuple._1);
}
});
final Long skewedUserid = reversedSampledRDD.sortByKey(false).take(1).get(0)._2;
// 从rdd1中分拆出导致数据倾斜的key形成独立的RDD。
JavaPairRDD skewedRDD = rdd1.filter(
new Function, Boolean>() {
private static final long serialVersionUID = 1L;
@Override
public Boolean call(Tuple2 tuple) throws Exception {
return tuple._1.equals(skewedUserid);
}
});
// 从rdd1中分拆出不导致数据倾斜的普通key形成独立的RDD。
JavaPairRDD commonRDD = rdd1.filter(
new Function, Boolean>() {
private static final long serialVersionUID = 1L;
@Override
public Boolean call(Tuple2 tuple) throws Exception {
return !tuple._1.equals(skewedUserid);
}
});
// rdd2就是那个所有key的分布相对较为均匀的rdd。
// 这里将rdd2中前面获取到的key对应的数据过滤出来分拆成单独的rdd并对rdd中的数据使用flatMap算子都扩容100倍。
// 对扩容的每条数据都打上0～100的前缀。
JavaPairRDD skewedRdd2 = rdd2.filter(
new Function, Boolean>() {
private static final long serialVersionUID = 1L;
@Override
public Boolean call(Tuple2 tuple) throws Exception {
return tuple._1.equals(skewedUserid);
}
}).flatMapToPair(new PairFlatMapFunction, String, Row>() {
private static final long serialVersionUID = 1L;
@Override
public Iterable> call(
Tuple2 tuple) throws Exception {
Random random = new Random();
List> list = new ArrayList>();
for(int i = 0; i < 100; i++) {
list.add(new Tuple2(i + "_" + tuple._1, tuple._2));
}
return list;
}
});
// 将rdd1中分拆出来的导致倾斜的key的独立rdd每条数据都打上100以内的随机前缀。
// 然后将这个rdd1中分拆出来的独立rdd与上面rdd2中分拆出来的独立rdd进行join。
JavaPairRDD> joinedRDD1 = skewedRDD.mapToPair(
new PairFunction, String, String>() {
private static final long serialVersionUID = 1L;
@Override
public Tuple2 call(Tuple2 tuple)
throws Exception {
Random random = new Random();
int prefix = random.nextInt(100);
return new Tuple2(prefix + "_" + tuple._1, tuple._2);
}
})
.join(skewedUserid2infoRDD)
.mapToPair(new PairFunction>, Long, Tuple2>() {
private static final long serialVersionUID = 1L;
@Override
public Tuple2> call(
Tuple2> tuple)
throws Exception {
long key = Long.valueOf(tuple._1.split("_")[1]);
return new Tuple2>(key, tuple._2);
}
});
// 将rdd1中分拆出来的包含普通key的独立rdd直接与rdd2进行join。
JavaPairRDD> joinedRDD2 = commonRDD.join(rdd2);
// 将倾斜key join后的结果与普通key join后的结果uinon起来。
// 就是最终的join结果。
JavaPairRDD> joinedRDD = joinedRDD1.union(joinedRDD2);
解决方案七使用随机前缀和扩容RDD进行join
方案适用场景如果在进行join操作时RDD中有大量的key导致数据倾斜那么进行分拆key也没什么意义此时就只能使用最后一种方案来解决问题了。
方案实现思路
1、该方案的实现思路基本和“解决方案六”类似首先查看RDD/Hive表中的数据分布情况找到那个造成数据倾斜的RDD/Hive表比如有多个key都对应了超过1万条数据。2、然后将该RDD的每条数据都打上一个n以内的随机前缀。3、同时对另外一个正常的RDD进行扩容将每条数据都扩容成n条数据扩容出来的每条数据都依次打上一个0~n的前缀。4、最后将两个处理后的RDD进行join即可。
方案实现原理将原先一样的key通过附加随机前缀变成不一样的key然后就可以将这些处理后的“不同key”分散到多个task中去处理而不是让一个task处理大量的相同key。该方案与“解决方案六”的不同之处就在于上一种方案是尽量只对少数倾斜key对应的数据进行特殊处理由于处理过程需要扩容RDD因此上一种方案扩容RDD后对内存的占用并不大而这一种方案是针对有大量倾斜key的情况没法将部分key拆分出来进行单独处理因此只能对整个RDD进行数据扩容对内存资源要求很高。
方案优点对join类型的数据倾斜基本都可以处理而且效果也相对比较显著性能提升效果非常不错。
方案缺点该方案更多的是缓解数据倾斜而不是彻底避免数据倾斜。而且需要对整个RDD进行扩容对内存资源要求很高。
方案实践经验曾经开发一个数据需求的时候发现一个join导致了数据倾斜。优化之前作业的执行时间大约是60分钟左右使用该方案优化之后执行时间缩短到10分钟左右性能提升了6倍。
// 首先将其中一个key分布相对较为均匀的RDD膨胀100倍。
JavaPairRDD expandedRDD = rdd1.flatMapToPair(
new PairFlatMapFunction, String, Row>() {
private static final long serialVersionUID = 1L;
@Override
public Iterable> call(Tuple2 tuple)
throws Exception {
List> list = new ArrayList>();
for(int i = 0; i < 100; i++) {
list.add(new Tuple2(0 + "_" + tuple._1, tuple._2));
}
return list;
}
});
// 其次将另一个有数据倾斜key的RDD每条数据都打上100以内的随机前缀。
JavaPairRDD mappedRDD = rdd2.mapToPair(
new PairFunction, String, String>() {
private static final long serialVersionUID = 1L;
@Override
public Tuple2 call(Tuple2 tuple)
throws Exception {
Random random = new Random();
int prefix = random.nextInt(100);
return new Tuple2(prefix + "_" + tuple._1, tuple._2);
}
});
// 将两个处理后的RDD进行join即可。
JavaPairRDD> joinedRDD = mappedRDD.join(expandedRDD);
解决方案八多种方案组合使用
在实践中发现很多情况下如果只是处理较为简单的数据倾斜场景那么使用上述方案中的某一种基本就可以解决。但是如果要处理一个较为复杂的数据倾斜场景那么可能需要将多种方案组合起来使用。比如说我们针对出现了多个数据倾斜环节的Spark作业可以先运用解决方案一和二预处理一部分数据并过滤一部分数据来缓解其次可以对某些shuffle操作提升并行度优化其性能最后还可以针对不同的聚合或join操作选择一种方案来优化其性能。大家需要对这些方案的思路和原理都透彻理解之后在实践中根据各种不同的情况灵活运用多种方案来解决自己的数据倾斜问题。
Spark性能优化Shuffle
调优篇
一、调优概述
大多数Spark作业的性能主要就是消耗在了shuffle环节因为该环节包含了大量的磁盘IO、序列化、网络数据传输等操作。因此如果要让作业的性能更上一层楼就有必要对shuffle过程进行调优。但是也必须提醒大家的是影响一个Spark作业性能的因素主要还是代码开发、资源参数以及数据倾斜shuffle调优只能在整个Spark的性能调优中占到一小部分而已。因此大家务必把握住调优的基本原则千万不要舍本逐末。下面我们就给大家详细讲解shuffle的原理以及相关参数的说明同时给出各个参数的调优建议。
二、ShuffleManager发展概述
在Spark的源码中负责shuffle过程的执行、计算和处理的组件主要就是ShuffleManager也即shuffle管理器。而随着Spark的版本的发展ShuffleManager也在不断迭代变得越来越先进。
在Spark 1.2以前默认的shuffle计算引擎是HashShuffleManager。该ShuffleManager而HashShuffleManager有着一个非常严重的弊端就是会产生大量的中间磁盘文件进而由大量的磁盘IO操作影响了性能。
因此在Spark 1.2以后的版本中默认的ShuffleManager改成了SortShuffleManager。SortShuffleManager相较于HashShuffleManager来说有了一定的改进。主要就在于每个Task在进行shuffle操作时虽然也会产生较多的临时磁盘文件但是最后会将所有的临时文件合并merge成一个磁盘文件因此每个Task就只有一个磁盘文件。在下一个stage的shuffle read task拉取自己的数据时只要根据索引读取每个磁盘文件中的部分数据即可。
下面我们详细分析一下HashShuffleManager和SortShuffleManager的原理。
2.1、HashShuffleManager运行原理
2.1.1、未经优化的HashShuffleManager
下图说明了未经优化的HashShuffleManager的原理。这里我们先明确一个假设前提每个Executor只有1个CPU core也就是说无论这个Executor上分配多少个task线程同一时间都只能执行一个task线程。
我们先从shuffle write开始说起。shuffle write阶段主要就是在一个stage结束计算之后为了下一个stage可以执行shuffle类的算子比如reduceByKey而将每个task处理的数据按key进行“分类”。所谓“分类”就是对相同的key执行hash算法从而将相同key都写入同一个磁盘文件中而每一个磁盘文件都只属于下游stage的一个task。在将数据写入磁盘之前会先将数据写入内存缓冲中当内存缓冲填满之后才会溢写到磁盘文件中去。
那么每个执行shuffle write的task要为下一个stage创建多少个磁盘文件呢很简单下一个stage的task有多少个当前stage的每个task就要创建多少份磁盘文件。比如下一个stage总共有100个task那么当前stage的每个task都要创建100份磁盘文件。如果当前stage有50个task总共有10个Executor每个Executor执行5个Task那么每个Executor上总共就要创建500个磁盘文件所有Executor上会创建5000个磁盘文件。由此可见未经优化的shuffle write操作所产生的磁盘文件的数量是极其惊人的。
接着我们来说说shuffle read。shuffle read通常就是一个stage刚开始时要做的事情。此时该stage的每一个task就需要将上一个stage的计算结果中的所有相同key从各个节点上通过网络都拉取到自己所在的节点上然后进行key的聚合或连接等操作。由于shuffle write的过程中task给下游stage的每个task都创建了一个磁盘文件因此shuffle read的过程中每个task只要从上游stage的所有task所在节点上拉取属于自己的那一个磁盘文件即可。
shuffle read的拉取过程是一边拉取一边进行聚合的。每个shuffle read task都会有一个自己的buffer缓冲每次都只能拉取与buffer缓冲相同大小的数据然后通过内存中的一个Map进行聚合等操作。聚合完一批数据后再拉取下一批数据并放到buffer缓冲中进行聚合操作。以此类推直到最后将所有数据到拉取完并得到最终的结果。
2.1.2、优化后的HashShuffleManager
下图说明了优化后的HashShuffleManager的原理。这里说的优化是指我们可以设置一个参数spark.shuffle.consolidateFiles。该参数默认值为false将其设置为true即可开启优化机制。通常来说如果我们使用HashShuffleManager那么都建议开启这个选项。
开启consolidate机制之后在shuffle write过程中task就不是为下游stage的每个task创建一个磁盘文件了。此时会出现shuffleFileGroup的概念每个shuffleFileGroup会对应一批磁盘文件磁盘文件的数量与下游stage的task数量是相同的。一个Executor上有多少个CPU core就可以并行执行多少个task。而第一批并行执行的每个task都会创建一个shuffleFileGroup并将数据写入对应的磁盘文件内。
当Executor的CPU core执行完一批task接着执行下一批task时下一批task就会复用之前已有的shuffleFileGroup包括其中的磁盘文件。也就是说此时task会将数据写入已有的磁盘文件中而不会写入新的磁盘文件中。因此consolidate机制允许不同的task复用同一批磁盘文件这样就可以有效将多个task的磁盘文件进行一定程度上的合并从而大幅度减少磁盘文件的数量进而提升shuffle write的性能。
假设第二个stage有100个task第一个stage有50个task总共还是有10个Executor每个Executor执行5个task。那么原本使用未经优化的HashShuffleManager时每个Executor会产生500个磁盘文件所有Executor会产生5000个磁盘文件的。但是此时经过优化之后每个Executor创建的磁盘文件的数量的计算公式为CPU core的数量 * 下一个stage的task数量。也就是说每个Executor此时只会创建100个磁盘文件所有Executor只会创建1000个磁盘文件。
2.2、SortShuffleManager运行原理
SortShuffleManager的运行机制主要分成两种一种是普通运行机制另一种是bypass运行机制。当shuffle read task的数量小于等于spark.shuffle.sort.bypassMergeThreshold参数的值时默认为200就会启用bypass机制。
2.2.1、普通运行机制
下图说明了普通的SortShuffleManager的原理。在该模式下数据会先写入一个内存数据结构中此时根据不同的shuffle算子可能选用不同的数据结构。如果是reduceByKey这种聚合类的shuffle算子那么会选用Map数据结构一边通过Map进行聚合一边写入内存如果是join这种普通的shuffle算子那么会选用Array数据结构直接写入内存。接着每写一条数据进入内存数据结构之后就会判断一下是否达到了某个临界阈值。如果达到临界阈值的话那么就会尝试将内存数据结构中的数据溢写到磁盘然后清空内存数据结构。
在溢写到磁盘文件之前会先根据key对内存数据结构中已有的数据进行排序。排序过后会分批将数据写入磁盘文件。默认的batch数量是10000条也就是说排序好的数据会以每批1万条数据的形式分批写入磁盘文件。写入磁盘文件是通过Java的BufferedOutputStream实现的。BufferedOutputStream是Java的缓冲输出流首先会将数据缓冲在内存中当内存缓冲满溢之后再一次写入磁盘文件中这样可以减少磁盘IO次数提升性能。
一个task将所有数据写入内存数据结构的过程中会发生多次磁盘溢写操作也就会产生多个临时文件。最后会将之前所有的临时磁盘文件都进行合并这就是merge过程此时会将之前所有临时磁盘文件中的数据读取出来然后依次写入最终的磁盘文件之中。此外由于一个task就只对应一个磁盘文件也就意味着该task为下游stage的task准备的数据都在这一个文件中因此还会单独写一份索引文件其中标识了下游各个task的数据在文件中的start offset与end offset。
SortShuffleManager由于有一个磁盘文件merge的过程因此大大减少了文件数量。比如第一个stage有50个task总共有10个Executor每个Executor执行5个task而第二个stage有100个task。由于每个task最终只有一个磁盘文件因此此时每个Executor上只有5个磁盘文件所有Executor只有50个磁盘文件。
2.2.2、bypass运行机制
下图说明了bypass SortShuffleManager的原理。bypass运行机制的触发条件如下
1、shuffle map task数量小于spark.shuffle.sort.bypassMergeThreshold参数的值。2、不是聚合类的shuffle算子比如reduceByKey。
此时task会为每个下游task都创建一个临时磁盘文件并将数据按key进行hash然后根据key的hash值将key写入对应的磁盘文件之中。当然写入磁盘文件时也是先写入内存缓冲缓冲写满之后再溢写到磁盘文件的。最后同样会将所有临时磁盘文件都合并成一个磁盘文件并创建一个单独的索引文件。
该过程的磁盘写机制其实跟未经优化的HashShuffleManager是一模一样的因为都要创建数量惊人的磁盘文件只是在最后会做一个磁盘文件的合并而已。因此少量的最终磁盘文件也让该机制相对未经优化的HashShuffleManager来说shuffle read的性能会更好。
而该机制与普通SortShuffleManager运行机制的不同在于第一磁盘写机制不同第二不会进行排序。也就是说启用该机制的最大好处在于shuffle write过程中不需要进行数据的排序操作也就节省掉了这部分的性能开销。
三、shuffle相关参数调优
以下是Shffule过程中的一些主要参数这里详细讲解了各个参数的功能、默认值以及基于实践经验给出的调优建议。
spark.shuffle.file.buffer
默认值32k参数说明该参数用于设置shuffle write task的BufferedOutputStream的buffer缓冲大小。将数据写到磁盘文件之前会先写入buffer缓冲中待缓冲写满之后才会溢写到磁盘。调优建议如果作业可用的内存资源较为充足的话可以适当增加这个参数的大小比如64k从而减少shuffle write过程中溢写磁盘文件的次数也就可以减少磁盘IO次数进而提升性能。在实践中发现合理调节该参数性能会有1%~5%的提升。
spark.reducer.maxSizeInFlight
默认值48m参数说明该参数用于设置shuffle read task的buffer缓冲大小而这个buffer缓冲决定了每次能够拉取多少数据。调优建议如果作业可用的内存资源较为充足的话可以适当增加这个参数的大小比如96m从而减少拉取数据的次数也就可以减少网络传输的次数进而提升性能。在实践中发现合理调节该参数性能会有1%~5%的提升。
spark.shuffle.io.maxRetries
默认值3参数说明shuffle read task从shuffle write task所在节点拉取属于自己的数据时如果因为网络异常导致拉取失败是会自动进行重试的。该参数就代表了可以重试的最大次数。如果在指定次数之内拉取还是没有成功就可能会导致作业执行失败。调优建议对于那些包含了特别耗时的shuffle操作的作业建议增加重试最大次数比如60次以避免由于JVM的full gc或者网络不稳定等因素导致的数据拉取失败。在实践中发现对于针对超大数据量数十亿~上百亿的shuffle过程调节该参数可以大幅度提升稳定性。
spark.shuffle.io.retryWait
默认值5s参数说明具体解释同上该参数代表了每次重试拉取数据的等待间隔默认是5s。调优建议建议加大间隔时长比如60s以增加shuffle操作的稳定性。
spark.shuffle.memoryFraction
默认值0.2参数说明该参数代表了Executor内存中分配给shuffle read task进行聚合操作的内存比例默认是20%。调优建议在资源参数调优中讲解过这个参数。如果内存充足而且很少使用持久化操作建议调高这个比例给shuffle read的聚合操作更多内存以避免由于内存不足导致聚合过程中频繁读写磁盘。在实践中发现合理调节该参数可以将性能提升10%左右。
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。
spark.shuffle.sort.bypassMergeThreshold
默认值200参数说明当ShuffleManager为SortShuffleManager时如果shuffle read task的数量小于这个阈值默认是200则shuffle write过程中不会进行排序操作而是直接按照未经优化的HashShuffleManager的方式去写数据但是最后会将每个task产生的所有临时磁盘文件都合并成一个文件并会创建单独的索引文件。调优建议当你使用SortShuffleManager时如果的确不需要排序操作那么建议将这个参数调大一些大于shuffle read task的数量。那么此时就会自动启用bypass机制map-side就不会进行排序了减少了排序的性能开销。但是这种方式下依然会产生大量的磁盘文件因此shuffle write性能有待提高。
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%。
写在最后的话
本文分别讲解了开发过程中的优化原则、运行前的资源参数设置调优、运行中的数据倾斜的解决方案、为了精益求精的shuffle调优。希望大家能够在阅读本文之后记住这些性能调优的原则以及方案在Spark作业开发、测试以及运行的过程中多尝试只有这样我们才能开发出更优的Spark作业不断提升其性能。
END