RDD算子是Spark計(jì)算框架中定義的對RDD進(jìn)行操作的各種函數(shù)，從RDD算子的功能可將RDD算子分為四類镜会，創(chuàng)建算子虑稼、轉(zhuǎn)換算子琳钉、緩存算子和行動算子。
RDD算子
創(chuàng)建算子
創(chuàng)建RDD有兩種方式：一種是將基于Scala的集合類型數(shù)據(jù)（如List或Set類型）分布到集群中生成RDD蛛倦，另一種則是加載外部數(shù)據(jù)源（如本地文本文件或HDFS文件）生成RDD歌懒。上面所提到的兩種方式都是通過SparkContext的接口函數(shù)提供的，前者有兩種方法：makeRDD和parallelize溯壶，后者則因?yàn)槠渲С植煌问胶筒煌袷降奈募霸恚休^多的函數(shù)。
基于集合類型數(shù)據(jù)創(chuàng)建RDD
SparkContext.makeRDD：創(chuàng)建RDD
# 輸入?yún)?shù)seq為一個(gè)集合數(shù)據(jù)集且改，參數(shù)String序列指定了希望將該數(shù)據(jù)集產(chǎn)生的RDD分區(qū)希望放置的節(jié)點(diǎn)验烧，
# 可以用Spark節(jié)點(diǎn)的主機(jī)名（hostname）描述
def makeRDD[T](seq: Seq[(T, Seq[String])])(implicit arg0: ClassTag[T]): [RDD]
# 輸入?yún)?shù)seq為一個(gè)數(shù)據(jù)集，numSlices是分區(qū)數(shù)量又跛，若不指定數(shù)量碍拆，
# 將使用Spark配置中的spark.default.parallelism參數(shù)所生成的defaultParallelism數(shù)值，為默認(rèn)的分區(qū)數(shù)量慨蓝。
def makeRDD[T](seq: Seq[T], numSlices: Int = [defaultParallelism])(implicit arg0: ClassTag[T]): [RDD]
makeRDD
示例代碼：
your-spark-path/bin# ./spark-shell
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel).
17/01/23 09:20:56 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
17/01/23 09:20:58 WARN spark.SparkContext: Use an existing SparkContext, some configuration may not take effect.
Spark context Web UI available at http://192.168.1.150:4040
Spark context available as 'sc' (master = local[*], app id = local-1485134457902).
Spark session available as 'spark'.
Welcome to
____ __
/ __/__ ___ _____/ /__
_\ \/ _ \/ _ `/ __/ '_/
/___/ .__/\_,_/_/ /_/\_\ version 2.0.2
/_/
Using Scala version 2.11.8 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_111)
Type in expressions to have them evaluated.
Type :help for more information.
scala> val rdd = sc.makeRDD(1 to 6, 2)
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at makeRDD at <console>:24
scala> rdd.collect
res0: Array[Int] = Array(1, 2, 3, 4, 5, 6)
scala> rdd.partitions
res1: Array[org.apache.spark.Partition] = Array(org.apache.spark.rdd.ParallelCollectionPartition@691, org.apache.spark.rdd.ParallelCollectionPartition@692)
scala> val data = Seq((1 to 6, Seq("spark-master", "hadoop-node1")), (7 to 10, Seq("hadoop-node2")))
data: Seq[(scala.collection.immutable.Range.Inclusive, Seq[String])] = List((Range(1, 2, 3, 4, 5, 6),List(spark-master, hadoop-node1)), (Range(7, 8, 9, 10),List(hadoop-node2)))
scala> val rdd1 = sc.makeRDD(data)
rdd1: org.apache.spark.rdd.RDD[scala.collection.immutable.Range.Inclusive] = ParallelCollectionRDD[1] at makeRDD at <console>:26
scala> rdd1.collect
res2: Array[scala.collection.immutable.Range.Inclusive] = Array(Range(1, 2, 3, 4, 5, 6), Range(7, 8, 9, 10))
scala> rdd1.preferredLocations(rdd1.partitions(1))
res6: Seq[String] = List(hadoop-node2)
scala> rdd1.preferredLocations(rdd1.partitions(0))
res7: Seq[String] = List(spark-master, hadoop-node1)
SparkContext.parallelize：數(shù)據(jù)并行化生成RDD
# 將集合數(shù)據(jù)seq分布到節(jié)點(diǎn)上形成RDD感混，并返回生成的RDD。numSlices是分區(qū)數(shù)量礼烈，
# 若不指定數(shù)量弧满，將使用Spark配置中的spark.default.parallelism參數(shù)所生成的defaultParallelism數(shù)值，為默認(rèn)的分區(qū)數(shù)量此熬。
def parallelize[T](seq: Seq[T], numSlices: Int = defaultParallelism)(implicit arg0: ClassTag[T]): RDD[T]
parallelize
scala> val rdd = sc.parallelize(1 to 6, 2)
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[2] at parallelize at <console>:24
scala> rdd.collect
res8: Array[Int] = Array(1, 2, 3, 4, 5, 6)
scala> rdd.partitions
res9: Array[org.apache.spark.Partition] = Array(org.apache.spark.rdd.ParallelCollectionPartition@6e3, org.apache.spark.rdd.ParallelCollectionPartition@6e4)
基于外部數(shù)據(jù)創(chuàng)建RDD
SparkContext.textFile——基于文本文件創(chuàng)建RDD
# 從HDFS庭呜、本地文件系統(tǒng)或者其他Hadoop支持的文件系統(tǒng)，按行讀入指定路徑下的文本文件摹迷，并返回生成的RDD疟赊。
# path是待讀入的文本文件的路徑，minPartitions是分區(qū)數(shù)量峡碉，不給定由spark配置中參數(shù)生成默認(rèn)值
def textFile(path: String, minPartitions: Int = defaultMinPartitions): RDD[String]
textFile
# 若讀取的數(shù)據(jù)是來自HDFS時(shí)，路徑地址："hdfs://...."
scala> val textFile = sc.textFile("file:/usr/local/spark/spark202/README.md")
textFile: org.apache.spark.rdd.RDD[String] = file:/usr/local/spark/spark202/README.md MapPartitionsRDD[6] at textFile at <console>:24
scala> textFile.count()
res11: Long = 99
scala> textFile.first()
res12: String = # Apache Spark
SparkContext.wholeTextFiles——基于一個(gè)目錄下的全部文本文件創(chuàng)建RDD
def wholeTextFiles(path: String, minPartitions: Int = defaultMinPartitions): RDD[(String, String)]
Read a directory of text files from HDFS, a local file system (available on all nodes), or any Hadoop-supported file system URI. Each file is read as a single record and returned in a key-value pair, where the key is the path of each file, the value is the content of each file.
基于Hadoop API從Hadoop文件數(shù)據(jù)創(chuàng)建RDD
Hadoop-RDD
