spark shuffle:分区原理及相关的疑问

Posted moonlightml

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了spark shuffle:分区原理及相关的疑问相关的知识,希望对你有一定的参考价值。

一、分区原理

1.为什么要分区?(这个借用别人的一段话来阐述。)

  为了减少网络传输,需要增加cpu计算负载。数据分区,在分布式集群里,网络通信的代价很大,减少网络传输可以极大提升性能。mapreduce框架的性能开支主要在io和网络传输,io因为要大量读写文件,它是不可避免的,但是网络传输是可以避免的,把大文件压缩变小文件, 从而减少网络传输,但是增加了cpu的计算负载。

Spark里面io也是不可避免的,但是网络传输spark里面进行了优化。spark把rdd进行分区(分片),放在集群上并行计算。同一个rdd分片100个,10个节点,平均一个节点10个分区,当进行sum型的计算的时候,先进行每个分区的sum然后把sum值shuffle传输到主程序进行全局sum,所以进行sum型计算对网络传输非常小。但对于进行join型的计算的时候,需要把数据本身进行shuffle,网络开销很大。

2.spark是如何优化这个问题的呢?

  • mapreduce分区:就是说mapreduce的网络传输主要在shuffle阶段,shuffle的根本原因是相同的key存在不同的节点上,按key进行聚合的时候不得不进行shuffle。shuffle是非常影响网络的,它要把所有的数据混在一起走网络,然后它才能把相同的key的数据拉到一起。要进行shuffle是存储决定的
  • spark分区:spark会基于key进行分区,也就是key的hashcode进行分区(相同的key,它的hashcode相同)。所以,它进行分区的时候100t的数据分成10份,每部分10个t,它能确保相同的key肯定在一个分区里面,而且它能保证存储的时候相同的key能够存在同一个节点上。例如,一个rdd分成了100份,集群有10个节点,所以每个节点存10份,每一份称为一个分区,spark能保证相同的key存在同一个节点上,实际上相同的key存在同一个分区。
  • key的分布不均决定了有的分区大有的分区小。没法保证分区数据量完全相等,但它会保证在一个接近的范围。所以对于mapreduce做的某些工作里边,spark就不需要shuffle了,spark解决网络传输这块的根本原理就是这个。

    进行join的时候是两个表,不可能把两个表都分区好,通常情况下是把用的频繁的大表事先进行分区,小表进行关联它的时候小表进行shuffle过程。

大表不需要shuffle。

   RDD 内部的数据集合在逻辑上(以及物理上)被划分成多个小集合,这样的每一个小集合被称为分区。像是下面这图中,三个 RDD,每个 RDD 内部都有两个分区。

 技术分享图片 

    在源码级别,RDD 类内存储一个 Partition 列表。每个 Partition 对象都包含一个 index 成员,通过 RDD 编号 + index 就能从唯一确定分区的 Block 编号,持久化的 RDD 就能通过这个 Block 编号从存储介质中获得对应的分区数据。(RDD + index-> Block 编号->分区数据)

二、Spark分区原则及方法

1.RDD分区的一个分区原则:尽可能是得分区的个数等于集群核心数目。

下面我们仅讨论Spark默认的分区个数,这里分别就parallelize和textFile具体分析其默认的分区数:

   无论是本地模式、Standalone模式、YARN模式或Mesos模式,我们都可以通过spark.default.parallelism来配置其默认分区个数,若没有设置该值,则根据不同的集群环境确定该值。

  • 本地模式:默认为本地机器的CPU数目,若设置了local[N],则默认为N
  • Apache Mesos:默认的分区数为8
  • Standalone或YARN:默认取集群中所有核心数目的总和,或者2,取二者的较大值。对于parallelize来说,没有在方法中的指定分区数,则默认为spark.default.parallelism,对于textFile来说,没有在方法中的指定分区数,则默认为min(defaultParallelism,2),而defaultParallelism对应的就是spark.default.parallelism。如果是从hdfs上面读取文件,其分区数为文件分片数(128MB/片)

2.如何创建分区?

有两种情况,创建 RDD 时和通过转换操作得到新 RDD 时。其实就是创建RDD的2种方法。

  • 对于前者,在调用 textFile 和 parallelize 方法时候手动指定分区个数即可。例如 sc.parallelize(Array(1, 2, 3, 5, 6), 2) 指定创建得到的 RDD 分区个数为 2。
  • 对于后者,直接调用 repartition 方法即可。实际上,分区的个数是根据转换操作对应多个 RDD 之间的依赖关系来确定的。

           1)窄依赖,子 RDD 由父 RDD 分区个数决定,例如 map 操作,父 RDD 和子 RDD 分区个数一致;

           2)Shuffle 依赖,则由分区器(Partitioner)决定,例如 groupByKey(new HashPartitioner(2)) 或者直接 groupByKey(2) 得到的新 RDD 分区个数等于 2。

3.spark shuffle中的partitioner的执行时机具体是哪里?

  •  partitioner在worker节点上执行,每个stage(除了最后一个)的最后一步就是将数据分区后,写入磁盘,然后把分区信息上报给master。
  • master在启动新的stage是会将上一个stage的分区信息携带给新的task,这样新的task就知道去哪读取数据了。

  分区是spark并行中的最小粒度,也就是说一个分区的数据必定需要一个线程来处理,不可拆分。几个分区就是并行几个task。

  例如,假定一个RDD的数据来源于2个hdfs文件,那么加载数据集时一开始默认是2个分区,可以并行处理2个文件。如果你有100个节点,每个节点有一个core,那么最多只能利用2个节点,这时我想提高并行度,就可以对这些数据进行重分区,把这两个文件分成100个分区,这时候就会使用hashpartitioner把这些数据散列到100个分区。

怎么做呢?

  因为一开始是2个分区,所以产生2个任务,分散到2个节点,每个任务各自利用hashpartitioner开始分区,分区完成的数据写入磁盘,这个时候2个节点的本地各自都会有100个分区的数据,编号0-99。也就是说一个分区的数据实际位于2个节点。然后它们把这些分区信息上报给driver,这样driver就知道这些分区的位置了。这个过程就是shuffle。

  接下来假设,我们需要对这个重分区后的RDD计数,此时就有100个分区,可以利用集群100个节点,。对每个分区来看,其实就是从driver上获取分区信息,然后从两个节点把该分区的数据通过网络捞出来,做累加计算(聚合reduce)。

 参考:

1.http://blog.csdn.net/jiangpeng59/article/details/52754928,Spark基础随笔:分区详解

2.http://blog.csdn.net/zengxiaosen/article/details/52637001-spark的优化-控制数据分区和分布

3.http://blog.csdn.net/jiangpeng59/article/details/52754928

以上是关于spark shuffle:分区原理及相关的疑问的主要内容,如果未能解决你的问题,请参考以下文章

[Spark性能调优] Spark Shuffle 中 JVM 内存使用及配置详情

[spark] shuffle

Spark Shuffle 机制解析

spark和mapreduce的shuffle

在同一个 Spark 作业中设置每个 shuffle 的 shuffle 分区数

spark.sql.shuffle.partitions 的 200 个默认分区难题