spark RDD算子详解1

Value型Transformation算子

处理数据类型为Value型的Transformation算子能够根据RDD变换算子的输入分区与输出分区关系分为如下几种类型。

1）输入分区与输出分区一对一型。

2）输入分区与输出分区多对一型。

3）输入分区与输出分区多对多型。

4）输出分区为输入分区子集型。

5）还有一种特殊的输入与输出分区一对一的算子类型：Cache型。Cache算子对RDD分区进行缓存。

这里的对应指的是分区依赖的对应

1．输入分区与输出分区一对一型

（1）map(func)

map是对RDD中的每一个元素都执行一个指定的函数来产生一个新的RDD，新RDD叫做MappedRDD(this, sc.clean(f))。任何原RDD中的元素在新RDD中都有且只有一个元素与之对应。

图3-4中的每一个方框表示一个RDD分区，左侧的分区通过用户自定义函数f:T->U映射为右侧的新的RDD分区。可是实际只有等到Action算子触发后，这个f函数才会和其余函数在一个Stage中对数据进行运算。V1输入f转换输出V’1。

（2）flatMap(func)

相似于map，可是每个输入元素，会被映射为0到多个输出元素（所以，func函数的返回值是一个Seq，而不是单一元素）。内部建立 FlatMappedRDD(this, sc.clean(f))。

图3-5中小方框表示RDD的一个分区，对分区进行flatMap函数操做，flatMap中传入的函数为f:T->U，T和U能够是任意的数据类型。将分区中的数据经过用户自定义函数f转换为新的数据。外部大方框能够认为是一个RDD分区，小方框表明一个集合。V一、V二、V3在一个集合做为RDD的一个数据项，转换为V’一、V’二、V’3后，将结合拆散，造成为RDD中的数据项。

（3）mapPartitions(func)

mapPartitions是map的一个变种。map的输入函数是应用于RDD中每一个元素，而mapPartitions的输入函数是应用于每一个分区，也就是把每一个分区中的内容做为总体来处理的。

mapPartitions函数获取到每一个分区的迭代器，在函数中经过这个分区总体的迭代器对整个分区的元素进行操做。内部实现是生成MapPartitionsRDD。图3-6中的方框表明一个RDD分区。

图3-6中，用户经过函数f (iter )=>iter.filter(_>=3)对分区中的全部数据进行过滤，>=3的数据保留。一个方块表明一个RDD分区，含有一、二、3的分区过滤只剩下元素3。

（4）glom()

glom函数将每一个分区造成一个数组，内部实现是返回的GlommedRDD。图3-7中的每一个方框表明一个RDD分区。

图3-7中的方框表明一个分区。该图表示含有V一、V二、V3的分区经过函数glom造成一个数组Array[(V1),(V2),(V3)]。

2．输入分区与输出分区多对一型

（1）union(otherDataset)

使用union函数时须要保证两个RDD元素的数据类型相同，返回的RDD数据类型和被合并的RDD元素数据类型相同，并不进行去重操做，保存全部元素。若是想去重，可使用distinct()。++符号至关于uion函数操做。

图3-8中左侧的大方框表明两个RDD，大方框内的小方框表明RDD的分区。右侧大方框表明合并后的RDD，大方框内的小方框表明分区。含有V1，V2…U4的RDD和含有V1，V8…U8的RDD合并全部元素造成一个RDD。V一、V一、V二、V8造成一个分区，其余元素同理进行合并。

（2）cartesian(otherDataset)

对两个RDD内的全部元素进行笛卡尔积操做。操做后，内部实现返回CartesianRDD。

左侧的大方框表明两个RDD，大方框内的小方框表明RDD的分区。右侧大方框表明合并后的RDD，大方框内的小方框表明分区。大方框表明RDD，大方框中的小方框表明RDD分区。 例如，V1和另外一个RDD中的W一、 W二、 Q5进行笛卡尔积运算造成（V1，W1）、（V1，W2）、（V1，Q5）。

3．输入分区与输出分区多对多型

groupBy (func)

将元素经过函数生成相应的Key，数据就转化为Key-Value格式，以后将Key相同的元素分为一组。
图中，方框表明一个RDD分区，相同key的元素合并到一个组。 例如，V1，V2合并为一个Key-Value对，其中key为“ V” ，Value为“ V1，V2” ，造成V，Seq（V1，V2）。

4．输出分区为输入分区子集型

（1）filter(func)

filter的功能是对元素进行过滤，对每一个元素应用f函数，返回值为true的元素在RDD中保留，返回为false的将过滤掉。内部实现至关于生成FilteredRDD(this，sc.clean(f))。
图3-11中的每一个方框表明一个RDD分区。T能够是任意的类型。经过用户自定义的过滤函数f，对每一个数据项进行操做，将知足条件，返回结果为true的数据项保留。例如，过滤掉V二、V3保留了V1，将区分命名为V1'。

（2）distinct([numTasks]))

distinct将RDD中的元素进行去重操做。图3-12中的方框表明RDD分区。

图3-12中的每一个方框表明一个分区，经过distinct函数，将数据去重。例如，重复数据V一、V1去重后只保留一份V1。

 

（3）subtract(other, numPartitions=None)

subtract至关于进行集合的差操做，RDD 1去除RDD 1和RDD 2交集中的全部元素。

图3-13中左侧的大方框表明两个RDD，大方框内的小方框表明RDD的分区。右侧大方框表明合并后的RDD，大方框内的小方框表明分区。V1在两个RDD中均有，根据差集运算规则，新RDD不保留，V2在第一个RDD有，第二个RDD没有，则在新RDD元素中包含V2。

（4）sample(withReplacement, fraction, seed=None)

sample将RDD这个集合内的元素进行采样，获取全部元素的子集。用户能够设定是否有放回的抽样、百分比、随机种子，进而决定采样方式。

内部实现是生成SampledRDD(withReplacement, fraction, seed)。

函数参数设置以下。

withReplacement=true，表示有放回的抽样；

withReplacement=false，表示无放回的抽样。

图3-14中的每一个方框是一个RDD分区。经过sample函数，采样50%的数据。V一、V二、U一、U二、U三、U4采样出数据V1和U一、U2，造成新的RDD。

（5）takeSample(withReplacement, num, seed=None)

takeSample()函数和上面的sample函数是一个原理，可是不使用相对比例采样，而是按设定的采样个数进行采样，同时返回结果再也不是RDD，而是至关于对采样后的数据进行Collect()，返回结果的集合为单机的数组。

图3-15中左侧的方框表明分布式的各个节点上的分区，右侧方框表明单机上返回的结果数组。经过takeSample对数据采样，设置为采样一份数据，返回结果为V1。

5．Cache型

（1）cache

cache将RDD元素从磁盘缓存到内存，至关于persist(MEMORY_ONLY)函数的功能。图3-14中的方框表明RDD分区。

图3-16中的每一个方框表明一个RDD分区，左侧至关于数据分区都存储在磁盘，经过cache算子将数据缓存在内存。

（2）persist(storageLevel=StorageLevel(False, True, False, False, 1))

persist函数对RDD进行缓存操做。数据缓存在哪里由StorageLevel枚举类型肯定。有如下几种类型的组合（见图3-15），DISK表明磁盘，MEMORY表明内存，SER表明数据是否进行序列化存储。

下面为函数定义，StorageLevel是枚举类型，表明存储模式，用户能够经过图3-17按需选择。

图3-17中列出persist函数能够缓存的模式。例如，MEMORY_AND_DISK_SER表明数据能够存储在内存和磁盘，而且以序列化的方式存储。其余同理。图中，方框表明RDD分区。 disk表明存储在磁盘，mem表明存储在内存。 数据最初所有存储在磁盘，经过persist（MEMORY_AND_DISK）将数据缓存到内存，可是有的分区没法容纳在内存，例如：图3-18中将含有V1，V2，V3的RDD存储到磁盘，将含有U1，U2的RDD仍旧存储在内存。