RDD行動算子：
行動算子就是會觸發action的算子，觸發action的含義就是真正的計算數據。
1. reduce
??函數簽名
def reduce(f: (T, T) => T): T
??函數說明
聚集?RDD 中的所有元素，先聚合分區內數據，再聚合分區間數據
val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4))
val reduceResult: Int = rdd.reduce(_+_)
println(reduceResult)
2. collect
??函數簽名
def collect(): Array[T]
??函數說明
在驅動程序中，以數組?Array 的形式返回數據集的所有元素
3. foreach
??函數簽名
def foreach(f: T => Unit): Unit = withScope {
?val cleanF = sc.clean(f)
?sc.runJob(this, (iter: Iterator[T]) => iter.foreach(cleanF))
}
??函數說明
分布式遍歷?RDD 中的每一個元素，調用指定函數
val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4))
rdd.collect().foreach(println)
4. count
??函數簽名
def count(): Long
??函數說明
返回?RDD 中元素的個數
val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4))
val countResult: Long = rdd.count()
println(countResult)
5. first
??函數簽名
def first(): T
??函數說明
返回 RDD 中的第一個元素
val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4))
val firstResult: Int = rdd.first()
println(firstResult)
6. take
??函數簽名
def take(num: Int): Array[T]
??函數說明
返回一個由?RDD 的前 n 個元素組成的數組
val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4))
val takeResult: Array[Int] = rdd.take(2)
takeResult.foreach(println)
7. takeOrdered
??函數簽名
def takeOrdered(num: Int)(implicit ord: Ordering[T]): Array[T]
??函數說明
返回該?RDD 排序后的前 n 個元素組成的數組
val rdd: RDD[Int] = sc.makeRDD(List(1,3,2,4))
val result: Array[Int] = rdd.takeOrdered(2)
result.foreach(println)
8. aggregate
??函數簽名
def aggregate[U: ClassTag](zeroValue: U)(seqOp: (U, T) => U, combOp: (U, U) => U): U
??函數說明
分區的數據通過初始值和分區內的數據進行聚合，然后再和初始值進行分區間的數據聚合
val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4),8)
// 將該?RDD 所有元素相加得到結果
val result1: Int = rdd.aggregate(0)(_+_, _+_)
val result2: Int = rdd.aggregate(10)(_+_,_+_)
println(result1)
println("**********")
9. fold
??函數簽名
def fold(zeroValue: T)(op: (T, T) => T): T
??函數說明
折疊操作，aggregate 的簡化版操作
val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4))
val foldResult: Int = rdd.fold(0)(_+_)
println(foldResult)
10. countByKey
??函數簽名
def countByKey(): Map[K, Long]
??函數說明
統計每種?key 的個數
val rdd: RDD[(Int, String)] = sc.makeRDD(List((1, "a"), (1, "a"), (1, "a"), (2,"b"), (3, "c"), (3, "c")))
val result: collection.Map[Int, Long] = rdd.countByKey()
print(result)
11. save 相關算子
??函數簽名
def saveAsTextFile(path: String): Unit
def saveAsObjectFile(path: String): Unit
def saveAsSequenceFile(
?path: String,
?codec: Option[Class[_ <: CompressionCodec]] = None): Unit //了解即可
??函數說明
將數據保存到不同格式的文件中
val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4))
// 保存成?Text 文件
rdd.saveAsTextFile("Spark-core/output/output")
// 序列化成對象保存到文件
rdd.saveAsObjectFile("Spark-core/output/output1")
——引用類型)
