Spark-RDD常用Transformationg与Action操作

一、概述

RDD创建后就可以在RDD上进行数据处理。RDD支持两种操作：转换（transformation），即从现有的数据集创建一个新的数据集；动作（action），即在数据集上进行计算后，返回一个值给Driver程序。

RDD 的转化操作是返回一个新的 RDD 的操作，比如 map() 和 filter() ，而行动操作则是向驱动器程序返回结果或把结果写入外部系统的操作，会触发实际的计算，比如 count() 和 first() 。Spark 对待转化操作和行动操作的方式很不一样，因此理解你正在进行的操作的类型是很重要的。如果对于一个特定的函数是属于转化操作还是行动操作感到困惑，你可以看看它的返回值类型：转化操作返回的是 RDD，而行动操作返回的是其他的数据类型。

RDD中所有的Transformation都是惰性的，也就是说，它们并不会直接计算结果。相反的它们只是记住了这些应用到基础数据集（例如一个文件）上的转换动作。只有当发生一个要求返回结果给Driver的Action时，这些Transformation才会真正运行。

这个设计让Spark更加有效的运行。

二、RDD支持的常用Transformation如下：

map(func)

返回一个新的分布式数据集，该数据集由每一个输入元素经过func函数转换后组成

fitler(func)

返回一个新的数据集，该数据集由经过func函数计算后返回值为true的输入元素组成

flatMap(func)

类似于map，但是每一个输入元素可以被映射为0或多个输出元素（因此func返回一个序列，而不是单一元素）

mapPartitions(func)

类似于map，但独立地在RDD上每一个分片上运行，因此在类型为T的RDD上运行时，func函数类型必须是Iterator[T]=>Iterator[U]

mapPartitionsWithSplit(func)

类似于mapPartitons，但func带有一个整数参数表示分片的索引值。因此在类型为T的RDD上运行时，func函数类型必须是(Int,Iterator[T])=>Iterator[U]

sample(withReplacement,fraction,seed)

根据fraction指定的比例对数据进行采样，可以选择是否用随机数进行替换，seed用于随机数生成器种子

union(otherDataSet)

返回一个新数据集，新数据集是由原数据集和参数数据集联合而成

distinct([numTasks])

返回一个包含原数据集中所有不重复元素的新数据集

groupByKey([numTasks])

在一个(K,V)数据集上调用，返回一个(K,Seq[V])对的数据集。注意默认情况下，只有8个并行任务来操作，但是可以传入一个可选的numTasks参数来改变它

reduceByKey(func,[numTasks])

在一个(K,V)对的数据集上调用，返回一个(K,V)对的数据集，使用指定的reduce函数，将相同的key的值聚合到一起。与groupByKey类似，reduceByKey任务的个数是可以通过第二个可选参数来设置的

sortByKey([[ascending],numTasks])

在一个(K,V)对的数据集上调用，K必须实现Ordered接口，返回一个按照Key进行排序的(K,V)对数据集。升序或降序由ascending布尔参数决定

join(otherDataset0,[numTasks])

在类型为(K,V)和(K,W)数据集上调用，返回一个相同的key对应的所有元素在一起的(K,(V,W))数据集

cogroup(otherDataset,[numTasks])

在类型为(K,V)和(K,W)数据集上调用，返回一个(K,Seq[V],Seq[W])元祖的数据集。这个操作也可以称为groupwith

cartesain(ohterDataset)

笛卡尔积，在类型为T和U类型的数据集上调用，返回一个(T,U)对数据集(两两的元素对)

三、RDD支持的常用action如下表

reduce(func)

通过函数func(接收两个参数，返回一个参数)聚集数据集中的所有元素。这个功能必须可交换且可关联的，从而可以正确的并行运行

collect()

在驱动程序中，以数组形式返回数据集中的所有元素。通常在使用filter或者其他操作返回一个足够小的数据子集后再使用会比较有用

count()

返回数据集元素个数

first()

返回数据集第一个元素(类似于take(1))

take(n)

返回一个由数据集前n个元素组成的数组

注意 这个操作目前并非并行执行，而是由驱动程序计算所有的元素

takeSample(withReplacement,num,seed)

返回一个数组，该数组由从数据集中随机采样的num个元素组成，可以选择是否由随机数替换不足的部分，seed用户指定随机数生成器种子

saveAsTextFile(path)

将数据集的元素以textfile的形式保存到本地文件系统—HDFS或者任何其他Hadoop支持的文件系统。对于每个元素，Spark将会调用toString方法，将它转换为文件中的文本行

saveAsSequenceFile(path)

将数据集中的元素以Hadoop sequencefile的格式保存到指定的目录下，可以是本地系统、HDFS或者任何其他的Hadoop支持的文件系统。这个只限于由key-value对组成，并实现了Hadoop的Writable接口，或者可以隐式的转换为Writable的RDD(Spark包括了基本类型转换，例如Int、Double、String等)

countByKey()

对(K,V)类型的RDD有效，返回一个(K,Int)对的map，表示每一个key对应的元素个数

foreach(func)

在数据集的每一个元素上，运行函数func进行更新。通常用于边缘效果，例如更新一个叠加器，或者和外部存储系统进行交互，如HBase