浅谈 Spark 应用程序的性能调优

标签: spark | 发表时间:2016-01-19 10:58 | 作者:青云QingCloud
出处:http://segmentfault.com/blogs

Spark是基于内存的分布式计算引擎,以处理的高效和稳定著称。然而在实际的应用开发过程中,开发者还是会遇到种种问题,其中一大类就是和性能相关。在本文中,笔者将结合自身实践,谈谈如何尽可能地提高应用程序性能。

分布式计算引擎在调优方面有四个主要关注方向,分别是CPU、内存、网络开销和I/O,其具体调优目标如下:

  • 提高CPU利用率。

  • 避免OOM。

  • 降低网络开销。

  • 减少I/O操作。

第1章 数据倾斜

数据倾斜意味着某一个或某几个Partition中的数据量特别的大,这意味着完成针对这几个Partition的计算需要耗费相当长的时间。

如果大量数据集中到某一个Partition,那么这个Partition在计算的时候就会成为瓶颈。图1是Spark应用程序执行并发的示意图,在Spark中,同一个应用程序的不同Stage是串行执行的,而同一Stage中的不同Task可以并发执行,Task数目由Partition数来决定,如果某一个Partition的数据量特别大,则相应的task完成时间会特别长,由此导致接下来的Stage无法开始,整个Job完成的时间就会非常长。

要避免数据倾斜的出现,一种方法就是选择合适的key,或者是自己定义相关的partitioner。在Spark中Block使用了ByteBuffer来存储数据,而ByteBuffer能够存储的最大数据量不超过2GB。如果某一个key有大量的数据,那么在调用cache或persist函数时就会碰到spark-1476这个异常。

下面列出的这些API会导致Shuffle操作,是数据倾斜可能发生的关键点所在

  1. groupByKey

  2. reduceByKey

  3. aggregateByKey

  4. sortByKey

  5. join

  6. cogroup

  7. cartesian

  8. coalesce

  9. repartition

  10. repartitionAndSortWithinPartitions


图1: Spark任务并发模型

  def rdd: RDD[T]
}

// TODO View bounds are deprecated, should use context bounds
// Might need to change ClassManifest for ClassTag in spark 1.0.0
case class DemoPairRDD[K <% Ordered[K] : ClassManifest, V: ClassManifest](
  rdd: RDD[(K, V)]) extends RDDWrapper[(K, V)] {
  // Here we use a single Long to try to ensure the sort is balanced, 
  // but for really large dataset, we may want to consider
  // using a tuple of many Longs or even a GUID
  def sortByKeyGrouped(numPartitions: Int): RDD[(K, V)] =
    rdd.map(kv => ((kv._1, Random.nextLong()), kv._2)).sortByKey()
    .grouped(numPartitions).map(t => (t._1._1, t._2))
}

case class DemoRDD[T: ClassManifest](rdd: RDD[T]) extends RDDWrapper[T] {
  def grouped(size: Int): RDD[T] = {
    // TODO Version where withIndex is cached
    val withIndex = rdd.mapPartitions(_.zipWithIndex)

    val startValues =
      withIndex.mapPartitionsWithIndex((i, iter) => 
        Iterator((i, iter.toIterable.last))).toArray().toList
      .sortBy(_._1).map(_._2._2.toLong).scan(-1L)(_ + _).map(_ + 1L)

    withIndex.mapPartitionsWithIndex((i, iter) => iter.map {
      case (value, index) => (startValues(i) + index.toLong, value)
    })
    .partitionBy(new Partitioner {
      def numPartitions: Int = size
      def getPartition(key: Any): Int = 
        (key.asInstanceOf[Long] * numPartitions.toLong / startValues.last).toInt
    })
    .map(_._2)
  }
}

定义隐式的转换

    implicit def toDemoRDD[T: ClassManifest](rdd: RDD[T]): DemoRDD[T] = 
    new DemoRDD[T](rdd)
  implicit def toDemoPairRDD[K <% Ordered[K] : ClassManifest, V: ClassManifest](
    rdd: RDD[(K, V)]): DemoPairRDD[K, V] = DemoPairRDD(rdd)
  implicit def toRDD[T](rdd: RDDWrapper[T]): RDD[T] = rdd.rdd
}

在spark-shell中就可以使用了

  import RDDConversions._

yourRdd.grouped(5)

第2章 减少网络通信开销

Spark的Shuffle过程非常消耗资源,Shuffle过程意味着在相应的计算节点,要先将计算结果存储到磁盘,后续的Stage需要将上一个Stage的结果再次读入。数据的写入和读取意味着Disk I/O操作,与内存操作相比,Disk I/O操作是非常低效的。

使用iostat来查看disk i/o的使用情况,disk i/o操作频繁一般会伴随着cpu load很高。

如果数据和计算节点都在同一台机器上,那么可以避免网络开销,否则还要加上相应的网络开销。 使用iftop来查看网络带宽使用情况,看哪几个节点之间有大量的网络传输。

图2是Spark节点间数据传输的示意图,Spark Task的计算函数是通过Akka通道由Driver发送到Executor上,而Shuffle的数据则是通过Netty网络接口来实现。由于Akka通道中参数spark.akka.framesize决定了能够传输消息的最大值,所以应该避免在Spark Task中引入超大的局部变量。


图2: Spark节点间的数据传输

第1节 选择合适的并发数

为了提高Spark应用程序的效率,尽可能的提升CPU的利用率。并发数应该是可用CPU物理核数的两倍。在这里,并发数过低,CPU得不到充分的利用,并发数过大,由于spark是每一个task都要分发到计算结点,所以任务启动的开销会上升。

并发数的修改,通过配置参数来改变spark.default.parallelism,如果是sql的话,可能通过修改spark.sql.shuffle.partitions来修改。

第1项 Repartition vs. Coalesce

repartition和coalesce都能实现数据分区的动态调整,但需要注意的是repartition会导致shuffle操作,而coalesce不会。

第2节 reduceByKey vs. groupBy

groupBy操作应该尽可能的避免,第一是有可能造成大量的网络开销,第二是可能导致OOM。以WordCount为例来演示reduceByKey和groupBy的差异

  reduceByKey
    sc.textFile(“README.md”).map(l=>l.split(“,”)).map(w=>(w,1)).reduceByKey(_ + _)


图3:reduceByKey的Shuffle过程

Shuffle过程如图2所示

  groupByKey
    sc.textFile(“README.md”).map(l=>l.split(“,”)).map(w=>(w,1)).groupByKey.map(r=>(r._1,r._2.sum))


图4:groupByKey的Shuffle过程

建议: 尽可能使用reduceByKey, aggregateByKey, foldByKey和combineByKey

假设有一RDD如下所示,求每个key的均值

  val data = sc.parallelize( List((0, 2.), (0, 4.), (1, 0.), (1, 10.), (1, 20.)) )

方法一:reduceByKey

  data.map(r=>(r._1, (r.2,1))).reduceByKey((a,b)=>(a._1 + b._1, a._2 + b._2)).map(r=>(r._1,(r._2._1/r._2._2)).foreach(println)

方法二:combineByKey

  data.combineByKey(value=>(value,1),
     (x:(Double, Int), value:Double)=> (x._1+value, x._2 + 1),     (x:(Double,Int), y:(Double, Int))=>(x._1 + y._1, x._2 + y._2))

第3节 BroadcastHashJoin vs. ShuffleHashJoin

在Join过程中,经常会遇到大表和小表的join. 为了提高效率可以使用BroadcastHashJoin, 预先将小表的内容广播到各个Executor, 这样将避免针对小表的Shuffle过程,从而极大的提高运行效率。

其实BroadCastHashJoin核心就是利用了BroadCast函数,如果理解清楚broadcast的优点,就能比较好的明白BroadcastHashJoin的优势所在。

以下是一个简单使用broadcast的示例程序。

  val lst = 1 to 100 toList
val exampleRDD = sc.makeRDD(1 to 20 toSeq, 2)
val broadcastLst = sc.broadcast(lst)
exampleRDD.filter(i=>broadcastLst.valuecontains(i)).collect.foreach(println)

第4节 map vs. mapPartitions

有时需要将计算结果存储到外部数据库,势必会建立到外部数据库的连接。应该尽可能的让更多的元素共享同一个数据连接而不是每一个元素的处理时都去建立数据库连接。

在这种情况下,mapPartitions和foreachPartitons将比map操作高效的多。

第5节 数据就地读取

移动计算的开销远远低于移动数据的开销。

Spark中每个Task都需要相应的输入数据,因此输入数据的位置对于Task的性能变得很重要。按照数据获取的速度来区分,由快到慢分别是:

  1. PROCESS_LOCAL

  2. NODE_LOCAL

  3. RACK_LOCAL

Spark在Task执行的时候会尽优先考虑最快的数据获取方式,如果想尽可能的在更多的机器上启动Task,那么可以通过调低spark.locality.wait的值来实现, 默认值是3s。

除了HDFS,Spark能够支持的数据源越来越多,如Cassandra, HBase,MongoDB等知名的NoSQL数据库,随着Elasticsearch的日渐兴起,spark和elasticsearch组合起来提供高速的查询解决方案也成为一种有益的尝试。

上述提到的外部数据源面临的一个相同问题就是如何让spark快速读取其中的数据, 尽可能的将计算结点和数据结点部署在一起是达到该目标的基本方法,比如在部署Hadoop集群的时候,可以将HDFS的DataNode和Spark Worker共享一台机器。

以cassandra为例,如果Spark的部署和Cassandra的机器有部分重叠,那么在读取Cassandra中数据的时候,通过调低spark.locality.wait就可以在没有部署Cassandra的机器上启动Spark Task。

对于Cassandra, 可以在部署Cassandra的机器上部署Spark Worker,需要注意的是Cassandra的compaction操作会极大的消耗CPU,因此在为Spark Worker配置CPU核数时,需要将这些因素综合在一起进行考虑。

这一部分的代码逻辑可以参考源码TaskSetManager::addPendingTask

  private def addPendingTask(index: Int, readding: Boolean = false) {
  // Utility method that adds `index` to a list only if readding=false or it's not already there
  def addTo(list: ArrayBuffer[Int]) {
    if (!readding || !list.contains(index)) {
      list += index
    }
  }

  for (loc <- tasks(index).preferredLocations) {
    loc match {
      case e: ExecutorCacheTaskLocation =>
        addTo(pendingTasksForExecutor.getOrElseUpdate(e.executorId, new ArrayBuffer))
      case e: HDFSCacheTaskLocation => {
        val exe = sched.getExecutorsAliveOnHost(loc.host)
        exe match {
          case Some(set) => {
            for (e <- set) {
              addTo(pendingTasksForExecutor.getOrElseUpdate(e, new ArrayBuffer))
            }
            logInfo(s"Pending task $index has a cached location at ${e.host} " +
              ", where there are executors " + set.mkString(","))
          }
          case None => logDebug(s"Pending task $index has a cached location at ${e.host} " +
              ", but there are no executors alive there.")
        }
      }
      case _ => Unit
    }
    addTo(pendingTasksForHost.getOrElseUpdate(loc.host, new ArrayBuffer))
    for (rack <- sched.getRackForHost(loc.host)) {
      addTo(pendingTasksForRack.getOrElseUpdate(rack, new ArrayBuffer))
    }
  }

  if (tasks(index).preferredLocations == Nil) {
    addTo(pendingTasksWithNoPrefs)
  }

  if (!readding) {
    allPendingTasks += index  // No point scanning this whole list to find the old task there
  }
}

如果准备让spark支持新的存储源,进而开发相应的RDD,与位置相关的部分就是自定义getPreferredLocations函数,以elasticsearch-hadoop中的EsRDD为例,其代码实现如下。

  override def getPreferredLocations(split: Partition): Seq[String] = {
  val esSplit = split.asInstanceOf[EsPartition]
  val ip = esSplit.esPartition.nodeIp
  if (ip != null) Seq(ip) else Nil
}

第6节 序列化

使用好的序列化算法能够提高运行速度,同时能够减少内存的使用。

Spark在Shuffle的时候要将数据先存储到磁盘中,存储的内容是经过序列化的。序列化的过程牵涉到两大基本考虑的因素,一是序列化的速度,二是序列化后内容所占用的大小。

kryoSerializer与默认的javaSerializer相比,在序列化速度和序列化结果的大小方面都具有极大的优势。所以建议在应用程序配置中使用KryoSerializer.

  spark.serializer  org.apache.spark.serializer.KryoSerializer

默认的cache没有对缓存的对象进行序列化,使用的StorageLevel是MEMORY_ONLY,这意味着要占用比较大的内存。可以通过指定persist中的参数来对缓存内容进行序列化。

  exampleRDD.persist(MEMORY_ONLY_SER)

需要特别指出的是persist函数是等到job执行的时候才会将数据缓存起来,属于延迟执行; 而unpersist函数则是立即执行,缓存会被立即清除。

更多内容可以访问  community.qingcloud.com

相关 [spark 应用程序 性能调优] 推荐:

浅谈 Spark 应用程序的性能调优

- - SegmentFault 最新的文章
Spark是基于内存的分布式计算引擎,以处理的高效和稳定著称. 然而在实际的应用开发过程中,开发者还是会遇到种种问题,其中一大类就是和性能相关. 在本文中,笔者将结合自身实践,谈谈如何尽可能地提高应用程序性能. 分布式计算引擎在调优方面有四个主要关注方向,分别是CPU、内存、网络开销和I/O,其具体调优目标如下:.

Spark性能调优

- - zzm
通常我们对一个系统进行性能优化无怪乎两个步骤——性能监控和参数调整,本文主要分享的也是这两方面内容. Spark提供了一些基本的Web监控页面,对于日常监控十分有用. http://master:4040(默认端口是4040,可以通过spark.ui.port修改)可获得这些信息:(1)stages和tasks调度情况;(2)RDD大小及内存使用;(3)系统环境信息;(4)正在执行的executor信息.

Spark的性能调优

- - 四火的唠叨
下面这些关于Spark的性能调优项,有的是来自官方的,有的是来自别的的工程师,有的则是我自己总结的. Data Serialization,默认使用的是Java Serialization,这个程序员最熟悉,但是性能、空间表现都比较差. 还有一个选项是Kryo Serialization,更快,压缩率也更高,但是并非支持任意类的序列化.

Spark&Spark性能调优实战

- - CSDN博客互联网推荐文章
       Spark特别适用于多次操作特定的数据,分mem-only和mem & disk. 其中mem-only:效率高,但占用大量的内存,成本很高;mem & disk:内存用完后,会自动向磁盘迁移,解决了内存不足的问题,却带来了数据的置换的消费. Spark常见的调优工具有nman、Jmeter和Jprofile,以下是Spark调优的一个实例分析:.

手把手教你 Spark 性能调优

- - ImportNew
上周四接到反馈,集群部分 spark 任务执行很慢,且经常出错,参数改来改去怎么都无法优化其性能和解决频繁随机报错的问题. 看了下任务的历史运行情况,平均时间 3h 左右,而且极其不稳定,偶尔还会报错:. 在有限的计算下,job的运行时长和数据量大小正相关,在本例中,数据量大小基本稳定,可以排除是日志量级波动导致的问题:.

HBase性能调优

- - 学着站在巨人的肩膀上
我们经常看到一些文章吹嘘某产品如何如何快,如何如何强,而自己测试时却不如描述的一些数据. 其实原因可能在于你还不是真正理解其内部结构,对于其性能调优方法不够了解. 本文转自TaoBao的Ken Wu同学的博客,是目前看到比较完整的HBase调优文章. 原文链接:HBase性能调优. 因官方Book Performance Tuning部分章节没有按配置项进行索引,不能达到快速查阅的效果.

hbase性能调优

- - 数据库 - ITeye博客
   1)、hbase.regionserver.handler.count:该设置决定了处理RPC的线程数量,默认值是10,通常可以调大,比如:150,当请求内容很大(上MB,比如大的put、使用缓存的scans)的时候,如果该值设置过大则会占用过多的内存,导致频繁的GC,或者出现OutOfMemory,因此该值不是越大越好.

Hadoop性能调优

- - 开源软件 - ITeye博客
是否对任务进行profiling,调用java内置的profile功能,打出相关性能信息. 对几个map或reduce进行profiling. 非常影响速度,建议在小数据量上尝试. 1表示不reuse,-1表示无限reuse,其他数值表示每个jvm reuse次数. reuse的时候,map结束时不会释放内存.

MapReduce - 性能调优

- - CSDN博客云计算推荐文章
        Hadoop为用户作业提供了多种可配置的参数,以允许用户根据作业特点调整这些参数值使作业运行效率达到最优.         对于一大批MapReduce程序,如果可以设置一个Combiner,那么对于提高作业性能是十分有帮助的. Combiner可减少Map Task中间输出的结果,从而减少各个Reduce Task的远程拷贝数据量,最终表现为Map Task和Reduce Task执行时间缩短.

Java 性能调优

- - 编程语言 - ITeye博客
1.用new关键词创建类的实例时,构造函数链中的所有构造函数都会被自动调用. 但如果一个对象实现了Cloneable接口,我们可以调用它的clone()方法. clone()方法不会调用任何类构造函数. 在使用设计模式(Design Pattern)的场合,如果用Factory模式创建对象,则改用clone()方法创建新的对象实例非常简单.