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

Stella981
• 阅读 748

一、分区的概念

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

二、为什么要进行分区

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

Spark里面io也是不可避免的,但是网络传输spark里面进行了优化:

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

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

  Spark把key-value rdd通过key的hashcode进行分区,而且保证相同的key存储在同一个节点上,这样对改rdd进行key聚合时,就不需要shuffle过程,我们进行mapreduce计算的时候为什么要进行shuffle?,就是说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。  

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

   分区是可配置的,只要RDD是基于键值对的即可

三、Spark分区原则及方法

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

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

3.1 本地模式

(1)默认方式

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

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

结果

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

(2)手动设置

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

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

结果

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

(3)跟local[n] 有关

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

如果n=* 那么分区个数就等于cpu core的个数

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

结果

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

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

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

(4)参数控制

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

结果

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

3.2 YARN模式

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

 进入defaultParallelism方法

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

继续进入defaultParallelism方法

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

这个一个trait,其实现类是(Ctrl+h)

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

进入TaskSchedulerImpl类找到defaultParallelism方法

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

继续进入defaultParallelism方法,又是一个trait,看其实现类

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

Ctrl+h看SchedulerBackend类的实现类

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

进入CoarseGrainedSchedulerBackend找到defaultParallelism

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

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

如果正常的情况下,那你设置了多少就是多少

四、分区器

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

    分区数我们是可以控制的,但是没必要有分区器。

(2)非key-value RDD分区,没必要设置分区器

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

  没必要设置,但是非要设置也行。

(3)Key-value形式的时候,我们就有必要了。

HashPartitioner

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

RangePartitioner

val resultRDD = testRDD.reduceByKey((x:Int,y:Int) => x+ y)
val newresultRDD=resultRDD.partitionBy(new RangePartitioner[String,Int](3,resultRDD))
println(newresultRDD.partitioner)
println("newresultRDD"+newresultRDD.getNumPartitions)

注:按照范围进行分区的,如果是字符串,那么就按字典顺序的范围划分。如果是数字,就按数据自的范围划分。

自定义分区

需要实现2个方法

class MyPartitoiner(val numParts:Int) extends  Partitioner{
  override def numPartitions: Int = numParts
  override def getPartition(key: Any): Int = {
    val domain = new URL(key.toString).getHost
    val code = (domain.hashCode % numParts)
    if (code < 0) {
      code + numParts
    } else {
      code
    }
  }
}

object DomainNamePartitioner {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("word count").setMaster("local")

    val sc = new SparkContext(conf)

    val urlRDD = 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)]]
    // = 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 = urlRDD.partitionBy(new HashPartitioner(2))
    hashPartitionedRDD.glom().collect()

    //使用spark-shell --jar的方式将这个partitioner所在的jar包引进去,然后测试下面的代码
    // spark-shell --master spark://master:7077 --jars spark-rdd-1.0-SNAPSHOT.jar
    val partitionedRDD = urlRDD.partitionBy(new MyPartitoiner(2))
    val array = partitionedRDD.glom().collect()

  }
}
点赞
收藏
评论区
推荐文章
blmius blmius
3年前
MySQL:[Err] 1292 - Incorrect datetime value: ‘0000-00-00 00:00:00‘ for column ‘CREATE_TIME‘ at row 1
文章目录问题用navicat导入数据时,报错:原因这是因为当前的MySQL不支持datetime为0的情况。解决修改sql\mode:sql\mode:SQLMode定义了MySQL应支持的SQL语法、数据校验等,这样可以更容易地在不同的环境中使用MySQL。全局s
皕杰报表之UUID
​在我们用皕杰报表工具设计填报报表时,如何在新增行里自动增加id呢?能新增整数排序id吗?目前可以在新增行里自动增加id,但只能用uuid函数增加UUID编码,不能新增整数排序id。uuid函数说明:获取一个UUID,可以在填报表中用来创建数据ID语法:uuid()或uuid(sep)参数说明:sep布尔值,生成的uuid中是否包含分隔符'',缺省为
Jacquelyn38 Jacquelyn38
3年前
2020年前端实用代码段,为你的工作保驾护航
有空的时候,自己总结了几个代码段,在开发中也经常使用,谢谢。1、使用解构获取json数据let jsonData  id: 1,status: "OK",data: 'a', 'b';let  id, status, data: number   jsonData;console.log(id, status, number )
Stella981 Stella981
3年前
Spark RDD工作原理详解+RDD JAVA API编程
第1章RDD概念1.1RDD为什么会产生   RDD:ResilientDistributedDataset弹性分布式数据集RDD是Spark的基石,是实现Spark数据处理的核心抽象。那么RDD为什么会产生呢?
Stella981 Stella981
3年前
Consistent hashing一致性算法原理
最近在整理redis分布式集群,首先就整理一下分布式算法原理。常见的分区规则有哈希分区和顺序分区两种,Redis采用的是哈希分区规则。节点取余分区使用特定的数据,如Redis的键或用户ID为key,节点数量为N,则:hash(key)%N,计算出哈希值,然后决定映射到哪个节点上,如节点数为4时,哈希值的结果可能为0、1、2,3.现假
Stella981 Stella981
3年前
Django中Admin中的一些参数配置
设置在列表中显示的字段,id为django模型默认的主键list_display('id','name','sex','profession','email','qq','phone','status','create_time')设置在列表可编辑字段list_editable
Wesley13 Wesley13
3年前
Mysql 表分区分类
针对Mysql数据库,表分区类型简析。【1】表分区类型(1)Range分区:按范围分区。按列值的范围区间进行分区存储;比如:id小于10存储在一个分区;id大于10小于20存储在另外一个分区;(2)List分区:按离散值集合分区。与range分区类似,不过它是按离散值进行分区。(3)Hash分区:按hash算法结果分区。对用户定义的表达式所返
Stella981 Stella981
3年前
Linux系统分区方式
硬盘在使用之前,要进行分区、格式化一个硬盘可以分主分区、扩展分区、逻辑分区    主分区扩展分区<4,可以全部设置成主分区,也可以主分区扩展分区进行设置,扩展分区不能直接使用,需要在扩展分区上建立逻辑分区,逻辑分区可以有多个  主分区扩展分区编号14  逻辑分区编号只能从5开始
Stella981 Stella981
3年前
Kafka重平衡机制
点击蓝色字体“肉眼品世界”,关注公众号深度价值体系传递!(https://oscimg.oschina.net/oscnet/cdaf2bb2b6804d68997f17d08fa4ea00.jpg)当集群中有新成员加入,或者某些主题增加了分区之后,消费者是怎么进行重新分配分区再进行消费的?这里就涉及到重平衡(Rebala
Python进阶者 Python进阶者
9个月前
Excel中这日期老是出来00:00:00,怎么用Pandas把这个去除
大家好,我是皮皮。一、前言前几天在Python白银交流群【上海新年人】问了一个Pandas数据筛选的问题。问题如下:这日期老是出来00:00:00,怎么把这个去除。二、实现过程后来【论草莓如何成为冻干莓】给了一个思路和代码如下:pd.toexcel之前把这