之前学过的kv类型上面的算子
groupby groupByKey reduceBykey sortBy sortByKey join[cogroup left inner right] shuffle的
mapValues keys values flatMapValues 普通算子，管道形式的算子
shuffle的过程是因为数据产生了打乱重分，分组、排序、join等算子需要将数据重新排版。
shuffle的过程是上游的数据处理完毕写出到自己的磁盘上，然后下游的数据从磁盘上面拉取。
重新排版打乱重分是需要存在规则的。
中间数据的流向规则叫做分区器 partitioner，这个分区器一般是存在于shuffle类算子中的，我们可以这么说，shuffle类算子一定会带有分区器，分区器也可以单独存在，人为定义分发规则。
groupBy groupBykey reduceBykey 自带的分区器HashPartitioner。
sortby sortBykey rangePartitioner
hashPartitioner
规则 按照key的hashCode %下游分区 = 分区编号
处理key-value类型数据，如果key为0，就分配去0号分区。否则调用nonNegativeMod函数。
保证取余的结果为正向结果。
hash取余的方式，不管数据分发到下游的什么分区中，最终结果都是相同的数据放入到一起。
演示结果：
scala> val arr = Array(1,2,3,4,5,6,7,8,9)
arr: Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8, 9)
scala> sc.makeRDD(arr,3)
res78: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[73] at makeRDD at <console>:27
scala> res78.mapPartitionsWithIndex((index,it)=> it.map((index,_)))
res79: org.apache.spark.rdd.RDD[(Int, Int)] = MapPartitionsRDD[74] at mapPartitionsWithIndex at <console>:26
scala> res79.collect
res80: Array[(Int, Int)] = Array((0,1), (0,2), (0,3), (1,4), (1,5), (1,6), (2,7), (2,8), (2,9))
scala> res78.map(t=>(t,t))
res81: org.apache.spark.rdd.RDD[(Int, Int)] = MapPartitionsRDD[75] at map at <console>:26
scala> res78.partitioner
res82: Option[org.apache.spark.Partitioner] = None
scala> res81.reduceByKey(_+_)
res84: org.apache.spark.rdd.RDD[(Int, Int)] = ShuffledRDD[76] at reduceByKey at <console>:26
scala> res84.partitioner
res85: Option[org.apache.spark.Partitioner] = Some(org.apache.spark.HashPartitioner@3)
scala> res84.mapPartitionsWithIndex((index,it)=> it.map((index,_)))
res88: org.apache.spark.rdd.RDD[(Int, (Int, Int))] = MapPartitionsRDD[77] at mapPartitionsWithIndex at <console>:26
scala> res88.collect
res89: Array[(Int, (Int, Int))] = Array((0,(6,6)), (0,(3,3)), (0,(9,9)), (1,(4,4)), (1,(1,1)), (1,(7,7)), (2,(8,8)), (2,(5,5)), (2,(2,2)))演示的逻辑，就是按照key.hashcode进行分区，int类型的hashcode值是自己的本身。
并且hash分区器的规则致使我们可以任意的修改下游的分区数量。
scala> res81.reduceByKey(_+_,100)
res91: org.apache.spark.rdd.RDD[(Int, Int)] = ShuffledRDD[78] at reduceByKey at <console>:26
scala> res91.partitions.size
res92: Int = 100
scala> res81.reduceByKey(_+_,2)
res93: org.apache.spark.rdd.RDD[(Int, Int)] = ShuffledRDD[79] at reduceByKey at <console>:26
scala> res93.partitions.size
res94: Int = 2rangePartitioner
hashPartitioner规则非常简单，直接规定来一个数据按照hashcode规则的分配，规则比较简答，但是会出现数据倾斜。
range分区规则中存在两个方法。
rangeBounds界限，在使用这个分区器之前先做一个界限划定。
首先使用水塘抽样算法，在未知的数据集中抽取能够代表整个数据集的样本，根据样本进行规则设定。
然后在使用getPartitions。
首先存在水塘抽样，规定数据的流向以后再执行整体逻辑，会先触发计算。
sortBykey是转换类的算子，不会触发计算。
但是我们发现它触发计算了，因为首先在计算之前先进行水塘抽样，能够规定下游的数据规则，然后再进行数据的计算。
scala> arr
res101: Array[Int] = Array(1, 9, 2, 8, 3, 7, 4, 6, 5)
scala> arr.map(t=> (t,t))
res102: Array[(Int, Int)] = Array((1,1), (9,9), (2,2), (8,8), (3,3), (7,7), (4,4), (6,6), (5,5))
scala> sc.makeRDD(res102)
res104: org.apache.spark.rdd.RDD[(Int, Int)] = ParallelCollectionRDD[94] at makeRDD at <console>:27
scala> res104.sortByKey()
res105: org.apache.spark.rdd.RDD[(Int, Int)] = ShuffledRDD[97] at sortByKey at <console>:26
scala> res105.partitioner
res106: Option[org.apache.spark.Partitioner] = Some(org.apache.spark.RangePartitioner@fe1f9dea)
scala> res104.sortByKey(true,3)
res107: org.apache.spark.rdd.RDD[(Int, Int)] = ShuffledRDD[100] at sortByKey at <console>:26
scala> res107.mapPartitionsWithIndex((index,it)=> it.map((index,_)))
res109: org.apache.spark.rdd.RDD[(Int, (Int, Int))] = MapPartitionsRDD[101] at mapPartitionsWithIndex at <console>:26
scala> res109.collect
res110: Array[(Int, (Int, Int))] = Array((0,(1,1)), (0,(2,2)), (0,(3,3)), (1,(4,4)), (1,(5,5)), (1,(6,6)), (2,(7,7)), (2,(8,8)), (2,(9,9)))range分区器，它是先做抽样然后指定下游分区的数据界限。
它可以修改分区数量，但是分区数量不能大于元素个数，必须要保证每个分区中都有元素。
scala> res104.sortByKey(true,3)
res111: org.apache.spark.rdd.RDD[(Int, Int)] = ShuffledRDD[104] at sortByKey at <console>:26
scala> res104.sortByKey(true,300)
res112: org.apache.spark.rdd.RDD[(Int, Int)] = ShuffledRDD[107] at sortByKey at <console>:26
scala> res111.partitions.size
res114: Int = 3
scala> res112.part自定义分区器
工作的过程中我们会遇见数据分类的情况，想要根据自己的需求定义分区的规则，让符合规则的数据发送到不同的分区中，这个时候我们就需要自定义分区器了。
定义分区器，让数据发送到不同的分区，从而不同的task任务输出的文件结果内容也不同
# 自己创建数据data/a.txt
hello tom hello jack
hello tom hello jack
hello tom hello jack
hello tom hello jack
hello tom hello jack
# 需求就是将数据按照规则进行分发到不同的分区中
# 存储的时候一个文件存储hello其他的文件存储tom jack分区器的定义需要实现分区器的接口
class MyPartitioner extends Partitioner{
override def numPartitions: Int = ???
// 设定下游存在几个分区
override def getPartition(key: Any): Int = ???
// 按照key设定分区的位置
}整体代码：
package com.hainiu.spark
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.rdd.RDD
import org.apache.spark.{Partitioner, SparkConf, SparkContext}
object Test1 {
def main(args: Array[String]): Unit = {
val conf = new SparkConf()
conf.setAppName("parse")
conf.setMaster("local[*]")
val sc = new SparkContext(conf)
val rdd = sc.textFile("data/a.txt")
.flatMap(_.split(" "))
.map((_, 1))
.reduceByKey(_ + _)
val rdd1 = rdd.partitionBy(new MyPartitioner)
val fs = FileSystem.get(new Configuration())
val out = "data/res"
if(fs.exists(new Path(out)))
fs.delete(new Path(out),true)
rdd1.saveAsTextFile(out)
}
}
class MyPartitioner extends Partitioner{
override def numPartitions: Int = 2
override def getPartition(key: Any): Int = {
if(key.toString.equals("hello"))
0
else
1
}
}
