文章目录
RDD行动算子
所谓的行动算子,其实不会再产生新的RDD,而是触发作业的执行。行动算子执行后,会获取到作业的执行结果。
转换算子不会触发作业的执行,只是功能的扩展和包装。
Spark的行动算子执行时,会产生Job对象,然后提交这个Job对象。
reduce
函数签名:
def reduce(f: (T, T) => T): T
函数说明:
聚集RDD中的所有元素,先聚合分区内数据,再聚合分区间数据
案例:
def main(args: Array[String]): Unit = {
val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD Action")
val sc = new SparkContext(sparkConf)
val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4))
val res: Int = rdd.reduce(_+_)
println(res) // 10
}
collect
函数签名:
def collect(): Array[T]
函数说明:
在driver中,以数组Array的形式返回数据集的所有元素。
注意:只有当结果数组预期很小时,才应使用此方法,因为所有数据都加载到driver的内存中。可能会出现内存溢出。
案例:
def main(args: Array[String]): Unit = {
val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD Action")
val sc = new SparkContext(sparkConf)
val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4))
val res: Array[Int] = rdd.collect()
println(res.mkString(", ")) // 1, 2, 3, 4
}
count
函数签名:
def count(): Long
函数说明:
返回RDD中元素的个数
案例:
def main(args: Array[String]): Unit = {
val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD Action")
val sc = new SparkContext(sparkConf)
val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4))
val res: Long = rdd.count() // 4
println(res)
}
countByValue
函数签名:
def countByValue()(implicit ord: Ordering[T] = null): Map[T,
Long]
函数说明:
将此RDD中每个唯一值的计数作为(value,count)对的本地map返回。
注意:只有当结果map预期很小时,才应使用此方法,因为所有的数据都会被加载到driver的内存中。
要处理非常大的结果,请考虑使用rdd.map(x => (x, 1L)).reduceByKey(_ + _),它返回一个RDD[T, Long],而不是一个map
案例:
def main(args: Array[String]): Unit = {
val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD Action")
val sc = new SparkContext(sparkConf)
val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4))
val res: collection.Map[Int, Long] = rdd.countByValue()
println(res.mkString(", ")) // 1 -> 1, 2 -> 1, 3 -> 1, 4 -> 1
}
first
函数签名:
def first(): T
函数说明:
返回RDD中的第一个元素
案例:
def main(args: Array[String]): Unit = {
val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD Action")
val sc = new SparkContext(sparkConf)
val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4))
val res: Int = rdd.first()
println(res) // 1
}
take
函数签名:
def take(num: Int): Array[T]
函数说明:
返回一个由RDD的前n个元素组成的数组。
取RDD的前num个元素。它首先扫描一个分区,然后使用该分区的结果来估计满足限制所需的附加分区的数量。
注意:只有当结果数组预期很小时,才应使用此方法,因为所有数据都加载到driver的内存中。
注意:由于内部实现的复杂性,如果对“Nothing”或“Null”的RDD调用此方法,则会引发异常。
案例:
def main(args: Array[String]): Unit = {
val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD Action")
val sc = new SparkContext(sparkConf)
val rdd: RDD[Int] = sc.makeRDD(List(4, 1, 3, 2))
val res: Array[Int] = rdd.take(3)
println(res.mkString(", ")) // 4, 1, 3
}
takeOrdered
函数签名:
def takeOrdered(num: Int)(implicit ord: Ordering[T]):
Array[T]
函数说明:
返回由指定的隐式Ordering[T]定义的该RDD中的前k个(最小的)元素,并维护该排序。这与[[top]]相反。
例如:
sc.parallelize(Seq(10, 4, 2, 12, 3)).takeOrdered(1)
// returns Array(2)
sc.parallelize(Seq(2, 3, 4, 5, 6)).takeOrdered(2)
// returns Array(2, 3)
注意:只有当结果数组预期很小时,才应使用此方法,因为所有数据都加载到driver的内存中。
案例:
def main(args: Array[String]): Unit = {
val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD Action")
val sc = new SparkContext(sparkConf)
val rdd: RDD[Int] = sc.makeRDD(List(4, 1, 3, 2))
val res: Array[Int] = rdd.takeOrdered(3)
println(res.mkString(", ")) // 1, 2, 3
}
aggregate
函数签名:
def aggregate[U: ClassTag](zeroValue: U)(seqOp: (U, T) => U,
combOp: (U, U) => U): U
函数说明:
分区的数据通过初始值和分区内的数据进行聚合,然后再和初始值进行分区间的数据集合。
使用给定的组合函数和一个中性的“zero value”,聚合每个分区的元素,然后聚合所有分区的结果。此函数会返回一个不同的结果类型U,与此RDD的类型T不同。因此,我们需要一个将T合并为U的操作和一个合并两个U的操作,就像使用scala.TraversableOnce。这两个函数都可以修改并返回它们的第一个参数,而不是创建一个新的U来避免内存的分配。
zeroValue:“seqOp”运算符的每个分区的累积结果的初始值,以及**“combOp”运算符的不同分区的合并结果的初始值**。
典型的中性元素:例如:Nil表示列表连接,或0表示求和
案例:
def main(args: Array[String]): Unit = {
val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD Action")
val sc = new SparkContext(sparkConf)
val rdd: RDD[Int] = sc.makeRDD(List(4, 1, 3, 2))
val res: Int = rdd.aggregate(10)(_+_, _+_)
println(res) // 100 // 因为local[*],八核
}
fold
函数签名:
def fold(zeroValue: T)(op: (T, T) => T): T
函数说明:
折叠操作,aggregate的简化版操作。
使用给定的关联函数和中性的“zero value”,聚合每个分区的元素,然后聚合所有分区的结果。允许函数op(t1,t2)修改t1并将其作为结果值返回,以避免对象分配;但是,它不应该修改t2。
这与Scala等函数语言中为非分布式集合实现的fold操作有些不同。此fold操作可以单独应用于分区,然后将这些结果fold到最终结果中,而不是按某些定义的顺序将fold应用于每个元素。对于不可交换的函数,结果可能与应用于非分布式集合的fold不同。
zeroValue:“op”运算符的每个分区的累积结果的初始值,以及“op”运算符的不同分区的合并结果的初始值
案例:
def main(args: Array[String]): Unit = {
val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD Action")
val sc = new SparkContext(sparkConf)
val rdd: RDD[Int] = sc.makeRDD(List(4, 1, 3, 2))
val res: Int = rdd.fold(10)(_+_)
println(res) // 100 // 因为local[*],八核
}
countByKey
函数签名:
def countByKey(): Map[K, Long]
函数说明:
计算每个key的元素数,将结果收集到本地map。
注意:只有当结果map预期很小时,才应使用此方法,因为整个过程都已加载到driver的内存中。要处理非常大的结果,请考虑使用rdd.mapValues(_ => 1L).reduceByKey(_ + _),它返回一个RDD[T,Long]而不是一个map。
案例:
def main(args: Array[String]): Unit = {
val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD Action")
val sc = new SparkContext(sparkConf)
val rdd: RDD[(Int, String)] = sc.makeRDD(List((1, "a"), (1, "a"), (1, "a"), (2, "b"), (3, "c"), (3, "c")))
val res: collection.Map[Int, Long] = rdd.countByKey()
println(res.mkString(", ")) // 1 -> 3, 2 -> 1, 3 -> 2
}
sava相关算子
函数签名&说明:
def saveAsTextFile(path: String): Unit
使用元素的字符串表示,将此RDD保存为文本文件。
def saveAsObjectFile(path: String): Unit
将此RDD另存为序列化对象的序列文件。
def saveAsSequenceFile( path: String, codec: Option[Class[_ <: CompressionCodec]] = None):
Unit
使用我们从RDD的键和值类型推断的可写类型,将RDD输出为Hadoop序列文件。如果键或值是可写的,则直接使用它们的类;否则,我们将原语类型(如Int和Double)映射为IntWritable、DoubleWritable等,将byte arrays映射到byteWritable,将Strings映射到文本。“path”可以在任何支持Hadoop的文件系统上。
案例:
def main(args: Array[String]): Unit = {
val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD Action")
val sc = new SparkContext(sparkConf)
val rdd: RDD[(Int, String)] = sc.makeRDD(List((1, "a"), (1, "a"), (1, "a"), (2, "b"), (3, "c"), (3, "c")))
rdd.saveAsTextFile("output")
rdd.saveAsObjectFile("output1")
rdd.saveAsSequenceFile("output3")
}
foreach
函数签名:
def foreach(f: T => Unit): Unit
函数说明:
分布式遍历RDD中的每一个元素,调用指定函数。
案例:
def main(args: Array[String]): Unit = {
val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD Action")
val sc = new SparkContext(sparkConf)
val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4))
// 收集后打印
// 集合的方法中的代码是在当前节点(Driver)中执行的。
// foreach方法是在当前节点的内存中完成数据的循环
rdd.collect().foreach(println)
/**
* 1
* 2
* 3
* 4
*/
println("--------------")
// 分布式打印
// rdd的方法称之为算子
// 算子的逻辑代码是在分布式计算节点Executor执行的
// foreach算子可以将循环在不同的计算节点完成
// 算子之外的代码是在Driver端执行
rdd.foreach(println)
/**
* 3
* 2
* 1
* 4
*/
}