| Action | 描述| | --- | --- | | `reduce(func)` | 使用函數 func(它接受兩個參數并返回一個)聚合數據集的元素。 | | `collect()` | 在驅動程序（Driver）中以數組的形式返回數據集的所有元素。 | | `count()` | 返回數據集中元素的數量。 | | `first()` | 返回數據集的第一個元素(類似于 take(1))。 | | `take(n)` | 返回一個包含數據集前 n 個元素的數組。 | | `takeSample(withReplacement, num, [seed])` | 返回一個數組，其中包含數據集的隨機 num 元素樣本，可以替換，也可以不替換，可以預先指定隨機數生成器種子。 | | `takeOrdered(n, [ordering])` | 使用 RDD 的自然順序或自定義比較器返回 RDD 的前 n 個元素。 | | `saveAsTextFile(path)` | 將數據集的元素作為文本文件(或文本文件集)寫入本地文件系統、HDFS 或任何其他 hadoop 支持的文件系統的給定目錄中。Spark 將對每個元素調用 toString，將其轉換為文件中的一行文本。 | | `saveAsSequenceFile(path)` | 將數據集的元素作為 Hadoop SequenceFile 寫入本地文件系統、HDFS 或任何其他 Hadoop 支持的文件系統的給定路徑中。這在實現 Hadoop 的可寫接口的鍵值對的 RDDs 上是可用的。在 Scala 中，它也可用于隱式轉換為可寫的類型(Spark 包括對Int、Double、String 等基本類型的轉換)。 | | `saveAsObjectFile(path)` | 使用 Java 序列化以簡單的格式編寫數據集的元素，然后可以使用 SparkContext.objectFile()加載這些元素。 | | `countByKey()` | 僅在類型(K, V)的 RDDs 上可用。返回(K, Int)對的Map 表示每個鍵的計數。 | | `foreach(func)` | 對數據集的每個元素運行函數 func | |`lookup(key)` | 用于PairRDD，返回K對應的所有V值 | 一些示例： ```scala import org.apache.spark.rdd.RDD import org.apache.spark.{SparkConf, SparkContext} object ActionOps { def main(args: Array[String]): Unit = { val conf:SparkConf = new SparkConf().setAppName(this.getClass.getName) .setMaster("local[4]") val sc:SparkContext = new SparkContext(conf) val rdd1:RDD[Int] = sc.parallelize(1 to 10) // 統計元素個數 println(rdd1.count()) // 10 // 收集RDD的所有數據，并遍歷輸出 rdd1.collect().foreach(x => print(s"$x ")) // 1 2 3 4 5 6 7 8 9 10 // 取出前5個元素 rdd1.take(5).foreach(x => print(s"$x ")) // 1 2 3 4 5 // 取出第一個元素 println(rdd1.first()) // 1 val words: RDD[String] = sc.parallelize(List("hello wrold hadoop spark ","spark python java hello")) println(words.flatMap(_.split("\\s+")).map((_, 1)).reduce((x, y) => (x._1, x._2 + y._2))) // (spark,8), 這里就是統計一共有多少個單詞 // 獲取最值與求和 println(rdd1.max()) // 10 println(rdd1.min) // 1 println(rdd1.sum()) // 55.0 // 將RDD輸出到文件中，文件目錄不能已經存在 // 如果要保持到hdfs上：hdfs://hadoop101:9000/data // 默認4個分區，所以會生成4個part-文件 words.saveAsTextFile("file:///E:\\hadoop\\output") // lookup val rdd2:RDD[(Char, Int)] = sc.parallelize(List(('a',1), ('a',2), ('b',3), ('c',4))) println(rdd2.lookup('a')) // WrappedArray(1, 2) sc.stop() // 釋放資源 } } ```