Spark RDD教程

这个教程将会帮助你理解和使用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") ``` >