Spark中repartition和coalesce的用法

Posted

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Spark中repartition和coalesce的用法相关的知识,希望对你有一定的参考价值。

参考技术A 在Spark的Rdd中,Rdd是分区的。

有时候需要重新设置Rdd的分区数量,比如Rdd的分区中,Rdd分区比较多,但是每个Rdd的数据量比较小,

每个任务计算的数据比较小时,计算速度有可能会变慢,因为处理的数据量小。但是任务的所需的调度时间会很多,

所以需要设置一个比较合理的分区。或者需要把Rdd的分区数量调大或是调小。

还有就是通过设置一个Rdd的分区来达到设置生成的文件的数量。

有两种方法是可以重设Rdd的分区:分别是 coalesce()方法和repartition()。

 这两个方法有什么区别,看看源码就知道了:

def coalesce(numPartitions: Int, shuffle: Boolean = false)(implicit ord: Ordering[T] = null)

    : RDD[T] = withScope

  if (shuffle)

    /** Distributes elements evenly across output partitions, starting from a random partition. */

    val distributePartition = (index: Int, items: Iterator[T]) =>

      var position = (new Random(index)).nextInt(numPartitions)

      items.map t =>

        // Note that the hash code of the key will just be the key itself. The HashPartitioner

        // will mod it with the number of total partitions.

        position = position + 1

        (position, t)

      

     : Iterator[(Int, T)]

    // include a shuffle step so that our upstream tasks are still distributed

    new CoalescedRDD(

      new ShuffledRDD[Int, T, T](mapPartitionsWithIndex(distributePartition),

      new HashPartitioner(numPartitions)),

      numPartitions).values

   else 

    new CoalescedRDD(this, numPartitions)

  



coalesce()方法的作用是返回指定一个新的指定分区的Rdd。

如果是生成一个窄依赖的结果,那么不会发生shuffle。比如:1000个分区被重新设置成10个分区,这样不会发生shuffle。

关于Rdd的依赖,这里提一下。Rdd的依赖分为两种:窄依赖和宽依赖。

窄依赖是指父Rdd的分区最多只能被一个子Rdd的分区所引用,即一个父Rdd的分区对应一个子Rdd的分区,或者多个父Rdd的分区对应一个子Rdd的分区。

而宽依赖就是宽依赖是指子RDD的分区依赖于父RDD的多个分区或所有分区,即存在一个父RDD的一个分区对应一个子RDD的多个分区。1个父RDD分区对应多个子RDD分区,这其中又分两种情况:1个父RDD对应所有子RDD分区(未经协同划分的Join)或者1个父RDD对应非全部的多个RDD分区(如groupByKey)。

如下图所示:map就是一种窄依赖,而join则会导致宽依赖

回到刚才的分区,如果分区的数量发生激烈的变化,如设置numPartitions = 1,这可能会造成运行计算的节点比你想象的要少,为了避免这个情况,可以设置shuffle=true,

那么这会增加shuffle操作。

关于这个分区的激烈的变化情况,比如分区数量从父Rdd的几千个分区设置成几个,有可能会遇到这么一个错误。

Exception in thread "main" org.apache.spark.SparkException: Job aborted due to stage failure: Task 1 in stage 77.0 failed 4 times, most recent failure: Lost task 1.3 in stage 77.0 (TID 6334, 192.168.8.61): java.io.IOException: Unable to acquire 16777216 bytes of memory

这个错误只要把shuffle设置成true即可解决。这个bug在spark1.6版本中被修复

当把父Rdd的分区数量增大时,比如Rdd的分区是100,设置成1000,如果shuffle为false,并不会起作用。

这时候就需要设置shuffle为true了,那么Rdd将在shuffle之后返回一个1000个分区的Rdd,数据分区方式默认是采用 hash partitioner。

最后来看看repartition()方法的源码:

def repartition(numPartitions: Int)(implicit ord: Ordering[T] =null): RDD[T] = withScope

    coalesce(numPartitions, shuffle =true)

 

从源码可以看出,repartition()方法就是coalesce()方法shuffle为true的情况。

如果是减少分区数量建议采用coalesce(numPartitions, false)方法,这样可以避免shuffle导致数据混洗,从而提高效率!

以上是关于Spark中repartition和coalesce的用法的主要内容,如果未能解决你的问题,请参考以下文章

Spark RDD 默认分区数量 - repartitions和coalesce异同

Spark 重分区函数:coalesce和repartition区别与实现,可以优化Spark程序性能

Spark transformation算子之coalesce&&repartition

Spark——算子之间的区别

coalesce和repartition的区别

coalesce