在Spark中转换算子并不会马上进行运算的，即所谓的“惰性运算”，而是在遇到行动算子时才会执行相应的语句的，触发Spark的任务调度开始进行计算。
Action-RDD
在这里我们可以将Spark中的行动算子分为两类：
- 1，数据运算类，主要用于触发RDD计算，并得到计算结果返回给Spark程序或Shell界面；
- 2，数据存储类，用于触发RDD计算后，将结果保存到外部存储系统中，如HDFS文件系统或数据库。
数据运算类行动算子
reduce——Reduce操作
def reduce(f: (T, T) => T): T
## Reduces the elements of this RDD using the specified commutative and associative binary operator.
reduce-RDD
scala> val nums = sc.parallelize(0 to 100, 4)
nums: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[54] at parallelize at <console>:25
scala> nums.reduce(_+_)
res39: Int = 5050
aggregate——聚合操作
aggregate操作使用参数seqOp指定的函数对每个分区里面的元素进行聚合，然后用参数combOp指定的函数将每个分区的聚合结果进行再次聚合，在进行combOp聚合时，计算的初始值由参数zeroValue指定。
def aggregate[U](zeroValue: U)(seqOp: (U, T) => U, combOp: (U, U) => U)(implicit arg0: ClassTag[U]): U
Paste_Image.png
scala> val rdd = sc.parallelize(1 to 6, 2)
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at <console>:24
scala> rdd.glom.collect
res0: Array[Array[Int]] = Array(Array(1, 2, 3), Array(4, 5, 6))
scala> rdd.aggregate(1)(_+_, Math.max(_,_))
res1: Int = 16
aggregate-RDD
collect——收集元素
collect作用是以数组格式返回RDD内的所有元素。
collectAsMap——收集Key/Value型RDD中的元素
收集Key/Value型RDD中的元素，并以Map数据类型的方式返回结果。
def collectAsMap(): Map[K, V]
collectAsMap-RDD
scala> val pairRDD = sc.parallelize(List((1,"wills"), (2, "april"), (3, "kris"), (4, "chang")),2)
pairRDD: org.apache.spark.rdd.RDD[(Int, String)] = ParallelCollectionRDD[2] at parallelize at <console>:24
scala> pairRDD.glom.collect
res3: Array[Array[(Int, String)]] = Array(Array((1,wills), (2,april)), Array((3,kris), (4,chang)))
scala> pairRDD.collectAsMap
res4: scala.collection.Map[Int,String] = Map(2 -> april, 4 -> chang, 1 -> wills, 3 -> kris)
count——计算元素个数
计算并返回RDD中元素的个数。
def count():Long
countByKey——按Key值统计Key/Value型RDD中的元素个数
计算Key/Value型RDD中每个Key值对应的元素个数，并以Map数据类型返回统计结果。
countByKey():Map[K, Long]
countByValue——统计RDD中元素值出现的次数
计算Key/Value型RDD中每个元素的值出现的次数，并以Map数据类型返回统计结果。
countByValue():Map[K, Long]
first——得到首个元素
first():T
glom——返回分区情况
原始RDD每个分区中的元素放到一个序列中，并汇集所有分区构成的序列生成新的RDD返回。
def glom():RDD[Array[T]]
fold——合并
将RDD中每个分区中的元素使用指定的函数参数f做合并计算，然后再加所有分区生成的结果使用函数f做合并。合并的初始值由zeroValue指定。
def fold(zeroValue: T)(op: (T, T) => T): T
fold-RDD
scala> val words = sc.parallelize(List("wills", "april", "kris", "chang"),2)
words: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[6] at parallelize at <console>:24
scala> words.glom.collect
res7: Array[Array[String]] = Array(Array(wills, april), Array(kris, chang))
scala> words.fold("&")(_+"##"+_)
res8: String = &##&##wills##april##&##kris##chang
fold-RDD
foreach——逐个处理RDD元素
使用参数f指定的函数对RDD中的每个元素进行相应的处理
def foreach(f: (T) => Unit): Unit
#Applies a function f to all elements of this RDD.
lookup——查找元素
在Key/Value型的RDD中，查找与参数key相同Key值的元素，并得到这些元素的Value值构成的序列。
def lookup(key:K):List[V]
scala> val pairs = sc.parallelize(List("apple", "banana", "berry", "cherry", "cumquat", "haw"), 1).keyBy(_.length)
pairs: org.apache.spark.rdd.RDD[(Int, String)] = MapPartitionsRDD[9] at keyBy at <console>:24
scala> pairs.collect
res9: Array[(Int, String)] = Array((5,apple), (6,banana), (5,berry), (6,cherry), (7,cumquat), (3,haw))
scala> pairs.lookup(5)
res10: Seq[String] = WrappedArray(apple, berry)
max——求最大值
def max()(implicit ord: Ordering[T]): T
min——求最小值
def min()(implicit ord: Ordering[T]): T
take——获取前n个元素
def take(num: Int): Array[T]
takeOrdered——获取排序后的前n个元素
def takeOrdered(num: Int)(implicit ord: Ordering[T]): Array[T]
takeSample——提取n个样本
def takeSample(withReplacement: Boolean, num: Int, seed: Long): Array[T]
top——寻找值最大的前几个元素
def top(num: Int)(implicit ord: Ordering[T]): Array[T]
Paste_Image.png
存储型行动算子
saveAsObjectFile——存储为二进制文件
def saveAsObjectFile(path: String): Unit
# Save this RDD as a SequenceFile of serialized objects.
saveAsTextFile——存储为文本文件
# Save this RDD as a text file, using string representations of elements.
def saveAsTextFile(path: String): Unit
