Spark——数据分区（进阶）

曾经终败给现在 2022-07-14 06:27 137阅读 0赞

对数据集在节点间的分区进行控制是Spark的一个特性之一。在分布式程序中通信的开销很大，和单节点的程序需要为记录集合选择合适的数据结构一样，Spark程序可以通过控制RDD分区方式来减少通信开销。只有当数据集多次在诸如连接这种基于键的操作中使用时，分区才会有帮助。若RDD只需要扫描一次，就没有必要进行分区处理。  
一、获取RDD的分区方式  
在Scala和java中，可以使用RDD的partitioner属性，java中使用partitioner()方法，他会返回一个scala.option对象，这是scala中用于存放可能存在的对象的容器类。对这个option调用isDefined来检查其中是否有值，调用get来获取其中的值，若存在值，这个值是一个spark.Partitioner对象。  
实例：

scala> val pairs=sc.parallelize(List((1,1),(2,2),(3,3)))
    pairs: org.apache.spark.rdd.RDD[(Int, Int)] = ParallelCollectionRDD[0] at parallelize at <console>:24
    
    scala> pairs.partitioner
    res0: Option[org.apache.spark.Partitioner] = None                                          ^
    
    scala> import org.apache.spark
    import org.apache.spark
    
    scala> val partitioned=pairs.partitionBy(new spark.HashPartitioner(2))
    partitioned: org.apache.spark.rdd.RDD[(Int, Int)] = ShuffledRDD[1] at partitionBy at <console>:27
    
    scala> partitioned.partitioner
    res1: Option[org.apache.spark.Partitioner] = Some(org.apache.spark.HashPartitioner@2)

可以在第四次输入时调用persist(),否则后续的RDD操作会对partitioned的整个谱系从新求职，这回导致对paids一遍又一遍的进行哈希分区操作。  
二、从分区中获益的操作  
能从spark分区中获取的操作有：cogroup()、groupWith()、join()、leftOuterJoin()、rightOuterJoin()、groupByKey()、reduceByKey()、combineByKey()以及lookup()。  
三、影响分区方式的操作  
Spark内部知道各操作影响分区方式，并将会对数据进行分区的操作的结果RDD自动设置对应的分区器。转化操作的结果并不一定按已知的分区方式分区，这时输出的RDD可能就会没有设置分区器。  
会为生成的结果RDD设好分区方式的操作：cogroup()、groupWith()、join()、leftOuterJoin()、rightOuterJoin()、groupByKey()、reduceByKey()、combineByKey()、partitionerBy()、sort()、mapValues()(若父RDD以有分区方式的话)、flatMapValues()(若父RDD有分区方式的话)、以及filter()(若父RDD有分区方式的话)。其他所有的操作生成的结果都不会存在特定的分区方式。  
四、自定义分区方式  
实现自定义的分区器，需要继承org.apache.spark.Partitioner类并实现下面三个方法：  
1、numPartitions:Int:返回创建出来的分区数  
2、getPartition():int:返回给定键的分区编号（0到numPartitions-1），确保永远返回的是一个非负数;  
3、equals():java判断相等的标准方法，这个方法的是吸纳非常重要，Spark需要用这个方法来检查你的分区器对象是否和其他分区器实例相同，这样spark才可以判断两个RDD的分区方式是否相同。