Spark系列--SparkCore(六)RDD分区详解

柔情只为你懂 2022-05-19 03:44 248阅读 0赞

转载自：[https://blog.csdn.net/jiangsanfeng1111/article/details/78191891][https_blog.csdn.net_jiangsanfeng1111_article_details_78191891]

## 一、分区个数规则 ##

spark.default.parallelism：（默认的并发数）= 2

当配置文件spark-default.conf中没有显示的配置，则按照如下规则取值：

**1、本地模式（不会启动executor，由SparkSubmit进程生成指定数量的线程数来并发）：**

spark-shell       spark.default.parallelism = 1
    
    spark-shell --master local[N] spark.default.parallelism = N （使用N个核）
    
    spark-shell --master local       spark.default.parallelism = 1

**2、伪集群模式（x为本机上启动的executor数，y为每个executor使用的core数，z为每个 executor使用的内存）**

spark-shell --master local-cluster[x,y,z] spark.default.parallelism = x * y

**3、其他模式（这里主要指yarn模式，当然standalone也是如此）**

Others: total number of cores on all executor nodes or 2, whichever is larger
    
    spark.default.parallelism =  max（所有executor使用的core总数， 2）

经过上面的规则，就能确定了spark.default.parallelism的默认值（前提是配置文件spark-default.conf中没有显示的配置，如果配置了，则spark.default.parallelism = 配置的值）

还有一个配置比较重要，spark.files.maxPartitionBytes = 128 M（默认）  
The maximum number of bytes to pack into a single partition when reading files.  
代表着rdd的一个分区能存放数据的最大字节数，如果一个400m的文件，只分了两个区，则在action时会发生错误。

当一个spark应用程序执行时，生成spark.context，同时会生成两个参数，由上面得到的spark.default.parallelism推导出这两个参数的值

sc.defaultParallelism     = spark.default.parallelism
    
    sc.defaultMinPartitions = min(spark.default.parallelism,2)

当sc.defaultParallelism和sc.defaultMinPartitions最终确认后，就可以推算rdd的分区数了。

--------------------

**产生rdd的几种方式：**

**1、通过scala 集合方式生成rdd**

> 如， val rdd = sc.parallelize(1 to 10)

这种方式下，如果在parallelize操作时没有指定分区数，则

> rdd的分区数 = sc.defaultParallelism

**2、通过textFile方式生成的rdd**

> 如， val rdd = sc.textFile(“path/file”)

有两种情况：

**a、从本地文件file:///生成的rdd，操作时如果没有指定分区数，则默认分区数规则为：**

> rdd的分区数 = max（本地file的分片数， sc.defaultMinPartitions）

**b、从hdfs分布式文件系统hdfs://生成的rdd，操作时如果没有指定分区数，则默认分区数规则为：**

> rdd的分区数 = max（hdfs文件的block数目， sc.defaultMinPartitions）

**3、从HBase的数据表转换为RDD，则该RDD的分区数为该Table的region数。**

String tableName ="pic_test2";
    conf.set(TableInputFormat.INPUT_TABLE,tableName);
    conf.set(TableInputFormat.SCAN,convertScanToString(scan));
    JavaPairRDD hBaseRDD = sc.newAPIHadoopRDD(conf,TableInputFormat.class,ImmutableBytesWritable.class,Result.class);
    Hbase Table:pic_test2的region为10，则hBaseRDD的分区数也为10。

**4、通过获取json（或者parquet等等）文件转换为DataFrame，则该DataFrame的分区数和该文件在文件系统中存放的Block数量对应。**

Dataset<Row> df = spark.read().json("examples/src/main/resources/people.json");
    people.json大小为300M，在HDFS中占用了3个blocks，则该DataFrame df分区数为3。

**5、Spark Streaming获取Kafka消息对应的分区数**

**a、基于Receiver接收数据**

在Receiver的方式中，Spark中的partition和kafka中的partition并不是相关的，所以如果我们加大每个topic的partition数量，

仅仅是增加线程来处理由单一Receiver消费的主题。但是这并没有增加Spark在处理数据上的并行度。

**b、基于direct直连方式读取kafka数据**

Spark会创建跟Kafka partition一样多的RDD partition，并且会并行从Kafka中读取数据。

所以在Kafka partition和RDD partition之间，有一个一对一的映射关系。

## 二、分区规则 ##

在Spark中分区器直接决定了RDD中分区的个数；也决定了RDD中每条数据经过Shuffle过程属于哪个分区；也决定了Reduce的个数。这三点看起来是不同的方面的，但其深层的含义是一致的。

*  我们需要注意的是，只有Key-Value类型的RDD才有分区的，非Key-Value类型的RDD分区的值是None的。
 *  在Spark中，存在两类分区函数：HashPartitioner和RangePartitioner，它们都是继承自Partitioner，主要提供了每个RDD有几个分区（numPartitions）以及对于给定的值返回一个分区ID（0~numPartitions-1），也就是决定这个值是属于那个分区的。

**HashPartitioner分区**

HashPartitioner分区的原理很简单，对于给定的key，如果key为null则分区ID为0，如果key不为null，计算其hashCode，并除于分区的个数取余，如果余数小于0，则用余数+分区的个数，最后返回的值就是这个key所属的分区ID。详细的查看HashPartitioner类源码。

**RangePartitioner分区**

从HashPartitioner分区的实现原理我们可以看出，其结果可能导致每个分区中数据量的不均匀，极端情况下会导致某些分区拥有RDD的全部数据，这显然不是我们需要的。而RangePartitioner分区则尽量保证每个分区中数据量的均匀，而且分区与分区之间是有序的，也就是说一个分区中的元素肯定都是比另一个分区内的元素小或者大；但是分区内的元素是不能保证顺序的。简单的说就是将一定范围内的数映射到某一个分区内。在实现中，分界的算法尤为重要。算法对应的函数是rangeBounds。使用的分区算法是：水塘抽样。

--------------------

## 三、自定义分区 ##

需要继承org.apache.spark.Partitioner类,实现如下：

class MySparkPartition(numParts: Int) extends Partitioner { 
    
      override def numPartitions: Int = numParts
    
      /** * 可以自定义分区算法 * @param key * @return */
      override def getPartition(key: Any): Int = {
        val domain = new java.net.URL(key.toString).getHost()
        val code = (domain.hashCode % numPartitions)
        if (code < 0) {
          code + numPartitions
        } else {
          code
        }
      }
      override def equals(other: Any): Boolean = other match {
        case mypartition: MySparkPartition =>
          mypartition.numPartitions == numPartitions
        case _ =>
          false
      }
      override def hashCode: Int = numPartitions
    
    }
    /** * def numPartitions：这个方法需要返回你想要创建分区的个数； * def getPartition：这个函数需要对输入的key做计算，然后返回该key的分区ID，范围一定是0到numPartitions-1； * equals()：这个是Java标准的判断相等的函数，之所以要求用户实现这个函数是因为Spark内部会比较两个RDD的分区是否一样，默认equals已经去区分两个RDD了 可以不重写 */

**使用自定义的partitioner对数据重新分区**

object UseMyPartitioner {
    
      def main(args: Array[String]) {
        val conf=new SparkConf()
          .setMaster("local[2]")
          .setAppName("TestMyParttioner")
          .set("spark.app.id","test-partition-id")
        val sc=new SparkContext(conf)
    
       //读取hdfs文件
        val lines=sc.textFile("hdfs://hadoop0:9000/test/word.txt")
        val splitMap=lines.flatMap(line=>line.split("\t")).map(word=>(word,1))//注意：RDD一定要是key-value 
    
       //保存
        splitMap.partitionBy(new MySparkPartition(3)).saveAsTextFile("F:/partrion/test")
    
        sc.stop()
    
      }
    
    }

[https_blog.csdn.net_jiangsanfeng1111_article_details_78191891]: https://blog.csdn.net/jiangsanfeng1111/article/details/78191891