Spark之RDD算子-行动算子

在Spark中转换算子并不会马上进行运算的,即所谓的“惰性运算”,而是在遇到行动算子时才会执行相应的语句的,触发Spark的任务调度开始进行计算。

《Spark之RDD算子-行动算子》 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.

《Spark之RDD算子-行动算子》 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

《Spark之RDD算子-行动算子》 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

《Spark之RDD算子-行动算子》 aggregate-RDD

collect——收集元素

collect作用是以数组格式返回RDD内的所有元素。

collectAsMap——收集Key/Value型RDD中的元素

收集Key/Value型RDD中的元素,并以Map数据类型的方式返回结果。

def collectAsMap(): Map[K, V]

《Spark之RDD算子-行动算子》 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

《Spark之RDD算子-行动算子》 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

《Spark之RDD算子-行动算子》 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]

《Spark之RDD算子-行动算子》 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

saveAsNewAPIHadoopFile——存储为Hadoop文件

saveAsNewAPIHadoopDataset——存储为Hadoop集

    原文作者:深思海数_willschang
    原文地址: https://www.jianshu.com/p/141da0f16963
    本文转自网络文章,转载此文章仅为分享知识,如有侵权,请联系博主进行删除。
点赞