RDD :弹性分布式数据集（分布式的元素集合）
Spark中，对数据的所有操作不外乎创建RDD,转化已有的RDD以及调用RDD操作进行求值，Spark会自动将RDD中的数据分发到集群上，并将操作并行化执行。
1、创建RDD
Spark有两种创建RDD的方式，一种是读取外部数据集，一种是在驱动器中对一个集合进行并行化。
最简单的创建RDD的方法就是讲程序中已有的一个集合传给SparkContext的parallelize()方法，这种方法可以快速的创建出RDD并对RDD进行操作，但是这种方法并不常用，方法如下：
val lines = sc.parallelize(List("pandas","i like pandas"))
常用的方法是从外部存储中读取数据来创建RDD，方法如下：
val lines = sc.textFile("/path/to/README.md")
2、RDD操作
RDD支持两种操作：转化操作和行动操作。
转化操作是会返回一个新的RDD的操作，比如map(),filter()，行动操作是向驱动器程序返回结果或者把结果写入外部系统的操作，会触发实际的计算,比如count(),first()。如果在实际操作中分不清自己的操作是什么操作，可以查看当前操作的返回值类型，转化操作返回的是RDD，行动操作返回的是其他类型。
转化操作 （是一种lazy操作）举例：filter()转化 下面程序展示了将log.txt中的error过滤出来。
val inputRDD = sc.textFile("log.txt") val errorRDD = inputRDD.filter(line => line.contains("error"))
行动操作 会把最终的结果返回到驱动程序，或者写入外部存储系统中，由于行动操作需要生成实际的输出，他们会强制执行那些求值必须用到的RDD的转化操作。
例：在scala中使用行动操作对错误进行计数
println("Input had " + badLinesRDD.count() + "concerning lines") println("Here are 10 examples:") badLinesRDD.take(10).foreach(println)
3、常见的转化操作和行动操作
基本RDD
最常用的转化操作是map()和filter()
map()接收一个函数，把这个函数用于RDD中的每一个元素，将函数的返回结果作为结果RDD中对应元素的值。
filter()接收一个函数，并将RDD中满足该函数的元素放入新的RDD中返回。
例： inputRDD = {1,2,3,4}
Mapped RDD (x*x) = {1,4,9,16}
Filtered RDD (x!=1) = {2,3,4}
代码举例 （计算平方）
val input = sc.parallelize(List(1,2,3,4)) val result = input.map(x => x*x) println(result.collect().mkString(","))
如果我们希望对每个输出元素生成多个输出元素，实现该功能的操作叫做flatMap()
使用flatMap将行数据切分为单词
val lines = sc.parallelize(List("hello world","hi")) val words = lines.flatMap(line => line.split(" ")) words.first() //"返回hello"
map 和 flatmap的区别举例
RDD1 = {"coffee panda","happy panda","happiest panda party"}
tokenize("coffee panda") = List("coffee","panda")
RDD1.map(tokenize) ={["coffee","panda"],["happy","panda"],["happiest","panda","party"]}
RDD1.flatMap(tokenize) ={"coffee","panda","happy","panda","happiest","panda","party"}
伪集合操作
RDD中最常缺失的集合属性是元素的唯一性，因为常常有重复的元素。如果只要唯一的元素，我们可以使用RDD.distinct()转化操作来生成一个只包含不同元素的新RDD。
行动操作
最常见的行动操作reduce()。接收一个函数作为参数，这个函数要操作两个相同元素类型的RDD数据并返回一个同样类型的新元素。举一个简历（函数+）
val sum = rdd.reduce((x,y) => x + y)
fold()函数的作用和reduce函数相同，但是需要提供一个初始值
rdd.fold(0)((x,y) => x+y)
4、持久化（缓存）
为了避免多次计算同一个RDD,可以让Spark对数据进行持久化，持久化存储一个RDD时，计算出RDD的节点会分别保存他们所求出的分区数据。如果一个有持久化的节点发生故障，Spark会在需要用到缓存的数据时重算丢失的数据分区。如果希望节点故障的情况下不会拖累我们的执行速度，也可以把数据备份到多个节点上。
使用persist()进行持久化操作
import org.apache.spark.storage.StorageLevel val result = input.map(x => x * x) result.persist(StorageLevel.DISK_ONLY) println(result.count()) println(result.collect().mkString(","))
还有一个方法unpersit()可以手动把持久化的RDD从缓存中移除。
