亲宝软件园·资讯

展开

Spark RDD教程

DCREN 人气:0
这个教程将会帮助你理解和使用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转换。 ![image](https://note.youdao.com/yws/public/resource/764cba932590c38a0a09266ad1ad6c56/xmlnote/WEBRESOURCE5b776f1b4938f691a6fb9f5dc6df3419/30680) 首先,从一个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") ``` >

加载全部内容

相关教程
猜你喜欢
用户评论