Spark RDD与Partion

一、RDD的概述

1.1　什么是RDD？

RDD（Resilient Distributed Dataset）叫做弹性分布式数据集，是Spark中最基本的数据抽象，它代表一个不可变、可分区、里面的元素可并行计算的集合。RDD具有数据流模型的特点：自动容错、位置感知性调度和可伸缩性。RDD允许用户在执行多个查询时显式地将工作集缓存在内存中，后续的查询能够重用工作集，这极大地提升了查询速度。

1.2　RDD的属性

（1）一组分片（Partition），即数据集的基本组成单位。对于RDD来说，每个分片都会被一个计算任务处理，并决定并行计算的粒度。用户可以在创建RDD时指定RDD的分片个数，如果没有指定，那么就会采用默认值。默认值就是程序所分配到的CPU Core的数目。

（2）一个计算每个分区的函数。Spark中RDD的计算是以分片为单位的，每个RDD都会实现compute函数以达到这个目的。compute函数会对迭代器进行复合，不需要保存每次计算的结果。

（3）RDD之间的依赖关系。RDD的每次转换都会生成一个新的RDD，所以RDD之间就会形成类似于流水线一样的前后依赖关系。在部分分区数据丢失时，Spark可以通过这个依赖关系重新计算丢失的分区数据，而不是对RDD的所有分区进行重新计算。

（4）一个Partitioner，即RDD的分片函数。当前Spark中实现了两种类型的分片函数，一个是基于哈希的HashPartitioner，另外一个是基于范围的RangePartitioner。只有对于于key-value的RDD，才会有Partitioner，非key-value的RDD的Parititioner的值是None。Partitioner函数不但决定了RDD本身的分片数量，也决定了parent RDD Shuffle输出时的分片数量。

（5）一个列表，存储存取每个Partition的优先位置（preferred location）。对于一个HDFS文件来说，这个列表保存的就是每个Partition所在的块的位置。按照“移动数据不如移动计算”的理念，Spark在进行任务调度的时候，会尽可能地将计算任务分配到其所要处理数据块的存储位置。

二、RDD编程API

Spark支持两个类型（算子）操作：Transformation和Action

2.1　Transformation

主要做的是就是将一个已有的RDD生成另外一个RDD。Transformation具有lazy特性(延迟加载)。Transformation算子的代码不会真正被执行。只有当我们的程序里面遇到一个action算子的时候，代码才会真正的被执行。这种设计让Spark更加有效率地运行。

常用的Transformation：

转换	含义
map(func)	返回一个新的RDD，该RDD由每一个输入元素经过func函数转换后组成
filter(func)	返回一个新的RDD，该RDD由经过func函数计算后返回值为true的输入元素组成
flatMap(func)	类似于map，但是每一个输入元素可以被映射为0或多个输出元素（所以func应该返回一个序列，而不是单一元素）
mapPartitions(func)	类似于map，但独立地在RDD的每一个分片上运行，因此在类型为T的RDD上运行时，func的函数类型必须是Iterator[T] => Iterator[U]
mapPartitionsWithIndex(func)	类似于mapPartitions，但func带有一个整数参数表示分片的索引值，因此在类型为T的RDD上运行时，func的函数类型必须是 (Int, Interator[T]) => Iterator[U]
sample(withReplacement, fraction, seed)	根据fraction指定的比例对数据进行采样，可以选择是否使用随机数进行替换，seed用于指定随机数生成器种子
union(otherDataset)	对源RDD和参数RDD求并集后返回一个新的RDD
intersection(otherDataset)	对源RDD和参数RDD求交集后返回一个新的RDD
distinct([numTasks]))	对源RDD进行去重后返回一个新的RDD
groupByKey([numTasks])	在一个(K,V)的RDD上调用，返回一个(K, Iterator[V])的RDD
reduceByKey(func, [numTasks])	在一个(K,V)的RDD上调用，返回一个(K,V)的RDD，使用指定的reduce函数，将相同key的值聚合到一起，与groupByKey类似，reduce任务的个数可以通过第二个可选的参数来设置
aggregateByKey(zeroValue)(seqOp, combOp, [numTasks])	先按分区聚合再总的聚合每次要跟初始值交流例如：aggregateByKey(0)(_+_,_+_) 对k/y的RDD进行操作
sortByKey([ascending], [numTasks])	在一个(K,V)的RDD上调用，K必须实现Ordered接口，返回一个按照key进行排序的(K,V)的RDD
sortBy(func,[ascending], [numTasks])	与sortByKey类似，但是更灵活第一个参数是根据什么排序第二个是怎么排序 false倒序第三个排序后分区数默认与原RDD一样
join(otherDataset, [numTasks])	在类型为(K,V)和(K,W)的RDD上调用，返回一个相同key对应的所有元素对在一起的(K,(V,W))的RDD 相当于内连接（求交集）
cogroup(otherDataset, [numTasks])	在类型为(K,V)和(K,W)的RDD上调用，返回一个(K,(Iterable<V>,Iterable<W>))类型的RDD
cartesian(otherDataset)	两个RDD的笛卡尔积的成很多个K/V
pipe(command, [envVars])	调用外部程序
coalesce(numPartitions)	重新分区第一个参数是要分多少区，第二个参数是否shuffle 默认false 少分区变多分区 true 多分区变少分区 false
repartition(numPartitions)	重新分区必须shuffle 参数是要分多少区少变多
repartitionAndSortWithinPartitions(partitioner)	重新分区+排序比先分区再排序效率高对K/V的RDD进行操作
foldByKey(zeroValue)(seqOp)	该函数用于K/V做折叠，合并处理，与aggregate类似第一个括号的参数应用于每个V值第二括号函数是聚合例如：_+_
combineByKey	合并相同的key的值 rdd1.combineByKey(x => x, (a: Int, b: Int) => a + b, (m: Int, n: Int) => m + n)
partitionBy（partitioner）	对RDD进行分区 partitioner是分区器例如new HashPartition(2
cache	RDD缓存，可以避免重复计算从而减少时间，区别：cache内部调用了persist算子，cache默认就一个缓存级别MEMORY-ONLY ，而persist则可以选择缓存级别
persist
Subtract（rdd）	返回前rdd元素不在后rdd的rdd
leftOuterJoin	leftOuterJoin类似于SQL中的左外关联left outer join，返回结果以前面的RDD为主，关联不上的记录为空。只能用于两个RDD之间的关联，如果要多个RDD关联，多关联几次即可。
rightOuterJoin	rightOuterJoin类似于SQL中的有外关联right outer join，返回结果以参数中的RDD为主，关联不上的记录为空。只能用于两个RDD之间的关联，如果要多个RDD关联，多关联几次即可
subtractByKey	substractByKey和基本转换操作中的subtract类似只不过这里是针对K的，返回在主RDD中出现，并且不在otherRDD中出现的元素

2.2　Action

触发代码的运行，我们一段spark代码里面至少需要有一个action操作。

常用的Action:

动作	含义
reduce(func)	通过func函数聚集RDD中的所有元素，这个功能必须是课交换且可并联的
collect()	在驱动程序中，以数组的形式返回数据集的所有元素
count()	返回RDD的元素个数
first()	返回RDD的第一个元素（类似于take(1)）
take(n)	返回一个由数据集的前n个元素组成的数组
takeSample(withReplacement,num, [seed])	返回一个数组，该数组由从数据集中随机采样的num个元素组成，可以选择是否用随机数替换不足的部分，seed用于指定随机数生成器种子
takeOrdered(n, [ordering])
saveAsTextFile(path)	将数据集的元素以textfile的形式保存到HDFS文件系统或者其他支持的文件系统，对于每个元素，Spark将会调用toString方法，将它装换为文件中的文本
saveAsSequenceFile(path)	将数据集中的元素以Hadoop sequencefile的格式保存到指定的目录下，可以使HDFS或者其他Hadoop支持的文件系统。
saveAsObjectFile(path)
countByKey()	针对(K,V)类型的RDD，返回一个(K,Int)的map，表示每一个key对应的元素个数。
foreach(func)	在数据集的每一个元素上，运行函数func进行更新。
aggregate	先对分区进行操作，在总体操作
reduceByKeyLocally
lookup
top
fold
foreachPartition

三、RDD Partition

3.1 详细介绍

我们要想对spark中RDD的分区进行一个简单的了解的话，就不免要先了解一下hdfs的前世今生。

众所周知，hdfs是一个非常不错的分布式文件系统，这是这么多年来大家有目共睹的。

hdfs文件为分布式存储，每个文件都被切分为block(默认为128M）。为了达到容错的目的，他们还提供为每个block存放了N个副本(默认为3个）。当然，以上说的这些也可以根据实际的环境业务调整。

多副本除了可以达到容错的目的，也为计算时数据的本地性提供了便捷。当数据所在节点的计算资源不充足时，多副本机制可以不用迁移数据，直接在另一个副本所在节点计算即可。此时看到这里，肯定就有人会问了，那如果所有副本所在的节点计算资源都不充足那该怎么办?

问的很好，一般会有一个配置来设置一个等待时长来等待的，假设等待时长为三秒，如果超过三秒，还没有空闲资源，就会分配给别的副本所在节点计算的，如果再别的副本所在节点也需等待且超过了三秒。则就会启动数据迁移了(诸多因素影响，代价就比较大了）。

接下来我们就介绍RDD，RDD是什么?弹性分布式数据集。

弹性:并不是指他可以动态扩展，而是血统容错机制。

分布式:顾名思义，RDD会在多个节点上存储，就和hdfs的分布式道理是一样的。hdfs文件被切分为多个block存储在各个节点上，而RDD是被切分为多个partition。不同的partition可能在不同的节点上。

再spark读取hdfs的场景下，spark把hdfs的block读到内存就会抽象为spark的partition。至于后续遇到shuffle的操作，RDD的partition可以根据Hash再次进行划分(一般pairRDD是使用key做Hash再取余来划分partition）。

再spark计算末尾，一般会把数据做持久化到hive，hbase，hdfs等等。我们就拿hdfs举例，将RDD持久化到hdfs上，RDD的每个partition就会存成一个文件，如果文件小于128M，就可以理解为一个partition对应hdfs的一个block。反之，如果大于128M，就会被且分为多个block，这样，一个partition就会对应多个block。

鉴于上述partition大于128M的情况，在做sparkStreaming增量数据累加时一定要记得调整RDD的分区数。假设，第一次保存RDD时10个partition，每个partition有140M。那么该RDD保存在hdfs上就会有20个block，下一批次重新读取hdfs上的这些数据，RDD的partition个数就会变为20个。再后续有类似union的操作，导致partition增加，但是程序有没有repartition或者进过shuffle的重新分区，这样就导致这部分数据的partition无限增加，这样一直下去肯定是会出问题的。所以，类似这样的情景，再程序开发结束一定要审查需不需要重新分区。

3.2 分区的3种方式

1、HashPartitioner

scala> val counts = sc.parallelize(List((1,'a'),(1,'aa'),(2,'b'),(2,'bb'),(3,'c')), 3)
.partitionBy(new HashPartitioner(3))

Spark中非常重要的一个分区器，也是默认分区器，默认用于90%以上的RDD相关API上；功能：依据RDD中key值的hashCode的值将数据取模后得到该key值对应的下一个RDD的分区id值，支持key值为null的情况，当key为null的时候，返回0；该分区器基本上适合所有RDD数据类型的数据进行分区操作；但是需要注意的是，由于JAVA中数组的hashCode是基于数组对象本身的，不是基于数组内容的，所以如果RDD的key是数组类型，那么可能导致数据内容一致的数据key没法分配到同一个RDD分区中，这个时候最好自定义数据分区器，采用数组内容进行分区或者将数组的内容转换为集合。

HashPartitioner确定分区的方式：partition = key.hashCode () % numPartitions

2、RangePartitioner

SparkCore中除了HashPartitioner分区器外，另外一个比较重要的已经实现的分区器，主要用于RDD的数据排序相关API中，比如sortByKey底层使用的数据分区器就是RangePartitioner分区器；该分区器的实现方式主要是通过两个步骤来实现的，第一步：先从整个RDD中抽取出样本数据，将样本数据排序，计算出每个分区的最大key值，形成一个Array[KEY]类型的数组变量rangeBounds；第二步：判断key在rangeBounds中所处的范围，给出该key值在下一个RDD中的分区id下标；该分区器要求RDD中的KEY类型必须是可以排序的。

3、CustomPartitioner

CustomPartitioner可以根据自己具体的应用需求，自定义分区。