rdd分片spark_Spark学习之路（十七）Spark分区

Spark学习之路 (十七)Spark分区

一、分区的概念

　　分区是RDD内部并行计算的一个计算单元&＃xff0c;RDD的数据集在逻辑上被划分为多个分片&＃xff0c;每一个分片称为分区&＃xff0c;分区的格式决定了并行计算的粒度&＃xff0c;而每个分区的数值计算都是在一个任务中进行的&＃xff0c;因此任务的个数&＃xff0c;也是由RDD(准确来说是作业最后一个RDD)的分区数决定。

二、为什么要进行分区

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

　　Spark里面io也是不可避免的&＃xff0c;但是网络传输spark里面进行了优化&＃xff1a;

　　Spark把rdd进行分区(分片)&＃xff0c;放在集群上并行计算。同一个rdd分片100个&＃xff0c;10个节点&＃xff0c;平均一个节点10个分区&＃xff0c;当进行sum型的计算的时候&＃xff0c;先进行每个分区的sum&＃xff0c;然后把sum值shuffle传输到主程序进行全局sum&＃xff0c;所以进行sum型计算对网络传输非常小。但对于进行join型的计算的时候&＃xff0c;需要把数据本身进行shuffle&＃xff0c;网络开销很大。

spark是如何优化这个问题的呢&＃xff1f;

　　Spark把key&＃xff0d;value rdd通过key的hashcode进行分区&＃xff0c;而且保证相同的key存储在同一个节点上&＃xff0c;这样对改rdd进行key聚合时&＃xff0c;就不需要shuffle过程&＃xff0c;我们进行mapreduce计算的时候为什么要进行shuffle&＃xff1f;&＃xff0c;就是说mapreduce里面网络传输主要在shuffle阶段&＃xff0c;shuffle的根本原因是相同的key存在不同的节点上&＃xff0c;按key进行聚合的时候不得不进行shuffle。shuffle是非常影响网络的&＃xff0c;它要把所有的数据混在一起走网络&＃xff0c;然后它才能把相同的key走到一起。要进行shuffle是存储决定的。

　　Spark从这个教训中得到启发&＃xff0c;spark会把key进行分区&＃xff0c;也就是key的hashcode进行分区&＃xff0c;相同的key&＃xff0c;hashcode肯定是一样的&＃xff0c;所以它进行分区的时候100t的数据分成10分&＃xff0c;每部分10个t&＃xff0c;它能确保相同的key肯定在一个分区里面&＃xff0c;而且它能保证存储的时候相同的key能够存在同一个节点上。比如一个rdd分成了100份&＃xff0c;集群有10个节点&＃xff0c;所以每个节点存10份&＃xff0c;每一分称为每个分区&＃xff0c;spark能保证相同的key存在同一个节点上&＃xff0c;实际上相同的key存在同一个分区。

　　key的分布不均决定了有的分区大有的分区小。没法分区保证完全相等&＃xff0c;但它会保证在一个接近的范围。所以mapreduce里面做的某些工作里边&＃xff0c;spark就不需要shuffle了&＃xff0c;spark解决网络传输这块的根本原理就是这个。

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

　　大表不需要shuffle。　　

　　需要在工作节点间进行数据混洗的转换极大地受益于分区。这样的转换是  cogroup&＃xff0c;groupWith&＃xff0c;join&＃xff0c;leftOuterJoin&＃xff0c;rightOuterJoin&＃xff0c;groupByKey&＃xff0c;reduceByKey&＃xff0c;combineByKey 和lookup。

　　 分区是可配置的&＃xff0c;只要RDD是基于键值对的即可。

三、Spark分区原则及方法

RDD分区的一个分区原则&＃xff1a;尽可能是得分区的个数等于集群核心数目

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

3.1　本地模式

(1)默认方式

以下这种默认方式就一个分区

结果

(2)手动设置

设置了几个分区就是几个分区

结果

(3)跟local[n] 有关

n等于几默认就是几个分区

如果n&＃61;* 那么分区个数就等于cpu core的个数

结果

本机电脑查看cpu core&＃xff0c;我的电脑--》右键管理--》设备管理器--》处理器

(4)参数控制

结果

3.2　YARN模式

 进入defaultParallelism方法

继续进入defaultParallelism方法

这个一个trait&＃xff0c;其实现类是(Ctrl&＃43;h)

进入TaskSchedulerImpl类找到defaultParallelism方法

继续进入defaultParallelism方法&＃xff0c;又是一个trait&＃xff0c;看其实现类

Ctrl&＃43;h看SchedulerBackend类的实现类

进入CoarseGrainedSchedulerBackend找到defaultParallelism

totalCoreCount.get()是所有executor使用的core总数&＃xff0c;和2比较去较大值

如果正常的情况下&＃xff0c;那你设置了多少就是多少

四、分区器

(1)如果是从HDFS里面读取出来的数据&＃xff0c;不需要分区器。因为HDFS本来就分好区了。

　　  分区数我们是可以控制的&＃xff0c;但是没必要有分区器。

(2)非key-value RDD分区&＃xff0c;没必要设置分区器

al testRDD &＃61; sc.textFile("C:\\Users\\Administrator\\IdeaProjects\\myspark\\src\\main\\hello.txt").flatMap(line &＃61;> line.split(",")).map(word &＃61;> (word, 1)).partitionBy(new HashPartitioner(2))

　　没必要设置&＃xff0c;但是非要设置也行。

(3)Key-value形式的时候&＃xff0c;我们就有必要了。

HashPartitioner

val resultRDD &＃61; testRDD.reduceByKey(new HashPartitioner(2),(x:Int,y:Int) &＃61;> x&＃43; y)//如果不设置默认也是HashPartitoiner&＃xff0c;分区数跟spark.default.parallelism一样println(resultRDD.partitioner)
println("resultRDD"&＃43;resultRDD.getNumPartitions)

RangePartitioner

val resultRDD &＃61; testRDD.reduceByKey((x:Int,y:Int) &＃61;> x&＃43; y)
val newresultRDD&＃61;resultRDD.partitionBy(new RangePartitioner[String,Int](3,resultRDD))
println(newresultRDD.partitioner)
println("newresultRDD"&＃43;newresultRDD.getNumPartitions)

注&＃xff1a;按照范围进行分区的&＃xff0c;如果是字符串&＃xff0c;那么就按字典顺序的范围划分。如果是数字&＃xff0c;就按数据自的范围划分。

自定义分区

需要实现2个方法

class MyPartitoiner(val numParts:Int) extends  Partitioner{  override def numPartitions: Int &＃61; numParts  override def getPartition(key: Any): Int &＃61; {val domain &＃61; new URL(key.toString).getHostval code &＃61; (domain.hashCode % numParts)    if (code <0) {code &＃43; numParts} else {code}}
}object DomainNamePartitioner {def main(args: Array[String]): Unit &＃61; {val conf &＃61; new SparkConf().setAppName("word count").setMaster("local")val sc &＃61; new SparkContext(conf)val urlRDD &＃61; sc.makeRDD(Seq(("http://baidu.com/test", 2),("http://baidu.com/index", 2), ("http://ali.com", 3), ("http://baidu.com/tmmmm", 4),("http://baidu.com/test", 4)))    //Array[Array[(String, Int)]]    // &＃61; Array(Array(),    // Array((http://baidu.com/index,2), (http://baidu.com/tmmmm,4),    // (http://baidu.com/test,4), (http://baidu.com/test,2), (http://ali.com,3)))val hashPartitionedRDD &＃61; urlRDD.partitionBy(new HashPartitioner(2))hashPartitionedRDD.glom().collect()    //使用spark-shell --jar的方式将这个partitioner所在的jar包引进去&＃xff0c;然后测试下面的代码    // spark-shell --master spark://master:7077 --jars spark-rdd-1.0-SNAPSHOT.jarval partitionedRDD &＃61; urlRDD.partitionBy(new MyPartitoiner(2))val array &＃61; partitionedRDD.glom().collect()}
}