这个教程将会帮助你理解和使用Apache Spark RDD。所有的在这个教程中使用的RDD例子将会提供在github上，供大家快速的浏览。
### 什么是RDD（Rssilient Distributed Dataset）？
RDD是Spark的基础数据结构，是Spark和Spark内核的主要数据抽象。RDD是容错的、不可变的对象分布式集合，这意味一旦创建了RDD，就不能更改它。RDD中的每个数据集都被划分为逻辑分区，这些逻辑分区可以在集群的不同节点上计算。
换句话说，RDD是类似于Scala中的集合的对象集合，不同之处在于RDD是分散在多个物理服务器（也称为集群中的节点）上的多个JVM上计算的，而Scala集合则位于单个JVM上。
另外，RDD提供对数据进行分区和分配的数据抽象，这些数据只在在多个节点上并行运行计算，而大多数时候，在RDD上进行转换时，我们不必担心默认情况下Spark提供的并行性。
本Apache Spark RDD教程使用Scala示例描述了RDD上可用的基本C座，例如map，filter和persist等。此外，本教程还介绍了pair RDD函数，该函数可在键值对的RDD上运行，例如groupByKey和join等。
### RDD的优势
- In-Memory Processing
- Immutability
- Fault Tolerance
- Lazy Evolution
- Partitioning
- Parallelize
### 限制
Spark RDDs不太适合对状态存储（如web应用程序的存储系统）进行更新的应用程序。对于这些应用程序，使用执行传统更新日志记录和数据检查点（如数据库）的系统更有效。RDD的目标是为批处理分析提供一个有效的编程模型，而不考虑这些异步应用程序。
### RDD的创建
RDD主要以两种不同的方式创建，首先是并行化现有集合，其次是引用外部存储系统（HDFS,S3等）中的数据集。
在查看实例之前，首先让我们使用SparkSession类中定义的builder模式方法初始化SparkSession。在初始化时，我们需要提供如下所示的主名称和应用程序名称。
```scala
val spark:SparkSession = SparkSession.builder()
.master("local[1]")
.appName("SparkByExamples.com")
.getOrCreate()
```
#### 使用sparkContext.parallelize()
sparkContext.parallelize用于并行化驱动程序中的现有集合。这是创建RDD的基本方法，主要在POC或原型制作时使用，它要求在创建RDD之前将所有数据都存在于驱动程序中，因此它并不是最常用于生产应用程序的。
```scala
val dataSeq = Seq(("Java", 1000), ("Python", 2000), ("Scala", 3000))
val rdd = spark.sparkContext.parallelize(dataSeq)
```
对于生产应用程序，我们主要通过使用外部存储系统(如HDFS、S3、HBase e.t.c)来创建RDD。
#### 使用sparkContext.textFile()
使用`testFile()`方法，我们能把一个txt文件读到RDD中。
```scala
val rdd2 = spark.sparkContext.textFile("/path/textFile.txt")
```
#### 使用sparkContext.wholeTextFiles()
`wholeTextFiles()`方法返回一个PairRDD，键是文件路径，值是内容
```scala
val rdd3 = spark.SparkContext.wholeTextFiles("/path/textFile.txt")
```
除了使用text文件，还可以使用csv文件，json和其他格式的文件。
#### 使用sparkContext.emptyRDD
使用sparkContext的emptyRDD()方法，创建一个没有数据的RDD。这个方法创建一个空的RDD，并且没有分区。
```scala
val rdd = spark.sparkContext.emptyRDD
val rddString = spark.sparkContext.emptyRDD[String]
```
#### 创建带分区的空的RDD
有时我们可能需要按分区将空的RDD写入文件，在这种情况下，您应该使用分区创建空的RDD。
```scala
val rdd2 = spark.sparkContext.parallelize(Seq.empty[String])
```
### RDD并行和重新分区
当我们使用parallelize()或textFile()或SparkContext的wholeTextFile()方法来初始化RDD时，它会根据资源可用性自动将数据分割为分区。
getNumPartitions- 返回数据的分区数。在RDD上应用的任何转换都是并行执行的。Spark将为集群的每个分区运行一个任务。
```scala
println("initial partition count:" + rdd.getNumPartitions)
// Outputs: initial partition count:2
```
手动设置并行度- 我们可以手动设置一个我们需要的分区数量，将分区数作为第二参数传递给这些函数`sparkContext.parallelize(dataSeq, 10)`
使用重新 分区和合并进行重新分配：有时候我们可能需要重新划分RDD，Spark提供了两种重新划分的方法；首先使用repartition()方法从所有节点shuffle数据，也称为完全混洗。第二种coalesce()方法，该方法shuffle最少节点的数据，举个例子，如果你有数据分布在4个分区，现在你使用coalesce(2)，仅仅只从两个节点移动数据。
这两个函数都会重新分配分区。`repartition()`方法的代价非常的巨大，它将会混洗集群上所有节点的数据。
```scala
val reparRdd = rdd.repartiton(4)
println("re-partition count:" + reparRdd.getNumPartitions)
// Outputs: "re-partition count:4"
```
Note:`repartition() or coalesce()`方法都返回一个新的RDD
### RDD操作
RDD转换：转换时惰性操作，这些操作不会更新RDD，而是返回另一个RDD
RDD操作：除法计算并返回RDD值得操作。
#### RDD转换例子
Spark RDD上的转换操作返回另一个RDD，并且转换操作是惰性的，这意味着他们不会立即执行，直到你调用一个RDD action时才会执行。RDD上的一些转换操作，如`flatMap, map, reductByKey, filter, sortByKey`，这些转换操作都会返回一个新的RDD，而不是更新已有的RDD。
在这个Spark RDD转换教程中，我将使用单词计数示例老解释转换。下图演示了我们将要使用的不同的RDD转换。
首先，从一个text文件创建一个RDD。
```scala
val rdd:RDD[String] = spark.spark.Context.textFile("src/main/scala/test.txt")
```
**flatMap**:`flatMap`转换将RDD展平，并返回新的RDD。在下面的示例中，首先它在RDD中空格分隔记录，最后将其展平。结果RDD在每个记录上都包含一个单词。
```scala
val rdd2 = rdd.flatMap(f => f.split(" "))
```
**map**:映射转换用于任何复杂的操作，比如添加一个列，更新一个列e.t.c。映射转换的输出总是与输入有相同数量的记录。
在我们的单词计数示例中，我们将为每个单词添加一个值为1的新列，RDD的结果为PairRDDFunctions，其中包含键值对，String类型的单词为Key，Int类型的1位为value。为了更好的理解，我们为rdd3变量定义了类型。
```scala
val rdd3:RDD[(String:Int)] = rdd2.map(m => (m, 1))
```
**filter**:filter转换操作是用来在RDD中过滤记录的。在我们的例子中，过滤所有以'a'开头的单词。
```scala
val rdd4 = rdd3.filter(a => a._1.startsWith("a"))
```
**reductByKey**:reduceByKey用指定的函数来合并相同key对应的value值。在我们的示例中，它通过对值应用sum函数来减少单词字符串。我们的RDD的结果包含唯一的单词和他们的计数。
```scala
val rdd5 = rdd4.reductByKey(_ + _)
```
**sortByKey**:sortByKey转换是对RDD的key列进行排序。在我们的示例中，首先我们使用映射转换RDD[(String,Int)] to RDD[(Int,String)]，并应用sortBykey，它在理想情况下对整数值进行排序。最后，使用println语句的foreach返回RDD中的所有单词及其作为键-值对的计数。
```scala
val rdd6 = rdd5.map(a => (a._2, a._1)).sortByKey()
// Print rd6 result to console
rdd6.foreach(println)
```
#### RDD Actions with example
RDD Action操作从RDD返回原始值。换句话说，任何返回非RDD[T]的RDD函数都被视为一个动作。
**count**:返回RDD中的记录数
```scala
//Action - count
println("Count : " + rdd6.count())
```
**first**:返回第一条记录
```scala
// Action - first
val firstRec = rdd6.first()
println("First Record : " + firstRec._1 + "," + firstRec._2)
```
**max**:返回最大的记录
```scala
val datMax = rdd6.max()
println("Max Record : " + datMax._1 + "," + datMax._2)
```
**reduct**:将记录减少为单个，我们可以使用它来计数或求和
```scala
val totalWordCount = rdd6.reduce((a, b) => (a._1 + b._1, a._2))
println("dataReduce Record : " + totalWordCount._1)
```
**take**:返回指定数目的记录
```scala
val data3 = rdd6.take(3)
data3.foreach(f => {
println("data3 Key:" + f._1 + ", Value:" + f._2)
})
```
**collect**:以数据形式返回RDD中的所有数据。当你在处理带有成千上万亿数据的巨大的RDD时，请小心使用此操作，因为你可能会耗尽驱动程序上的内存。
```scala
val data = rdd6.collect()
data.foreach(f => {
println("Key:" + f._1 + ", Value:" + f._2)
})
```
**saveAsTextFile**:使用saveAsTextFile操作，可以把RDD写入到text文件。
```scala
rdd6.saveAsTextFile("/tmp/wordCount")
```
>
