Spark RDD怎么创建

Spark RDD怎么创建

这篇文章主要介绍“ Spark RDD怎么创建”，在日常操作中，相信很多人在 Spark RDD怎么创建问题上存在疑惑，小编查阅了各式资料，整理出简单好用的操作方法，希望对大家解答” Spark RDD怎么创建”的疑惑有所帮助！接下来，请跟着小编一起来学习吧！

一、RDD的概述

1.1　什么是RDD？

RDD（Resilient Distributed Dataset）叫做弹性分布式数据集，是Spark中最基本的数据抽象，它代表一个不可变、可分区、里面的元素可并行计算的集合。RDD具有数据流模型的特点：自动容错、位置感知性调度和可伸缩性。RDD允许用户在执行多个查询时显式地将工作集缓存在内存中，后续的查询能够重用工作集，这极大地提升了查询速度。

1.2　RDD的属性

（1）一组分片（Partition），即数据集的基本组成单位。对于RDD来说，每个分片都会被一个计算任务处理，并决定并行计算的粒度。用户可以在创建RDD时指定RDD的分片个数，如果没有指定，那么就会采用默认值。默认值就是程序所分配到的CPU Core的数目。

（2）一个计算每个分区的函数。Spark中RDD的计算是以分片为单位的，每个RDD都会实现compute函数以达到这个目的。compute函数会对迭代器进行复合，不需要保存每次计算的结果。

（3）RDD之间的依赖关系。RDD的每次转换都会生成一个新的RDD，所以RDD之间就会形成类似于流水线一样的前后依赖关系。在部分分区数据丢失时，Spark可以通过这个依赖关系重新计算丢失的分区数据，而不是对RDD的所有分区进行重新计算。

（4）一个Partitioner，即RDD的分片函数。当前Spark中实现了两种类型的分片函数，一个是基于哈希的HashPartitioner，另外一个是基于范围的RangePartitioner。只有对于于key-value的RDD，才会有Partitioner，非key-value的RDD的Parititioner的值是None。Partitioner函数不但决定了RDD本身的分片数量，也决定了parent RDD Shuffle输出时的分片数量。

（5）一个列表，存储存取每个Partition的优先位置（preferred location）。对于一个HDFS文件来说，这个列表保存的就是每个Partition所在的块的位置。按照“移动数据不如移动计算”的理念，Spark在进行任务调度的时候，会尽可能地将计算任务分配到其所要处理数据块的存储位置。

1.3　WordCount粗图解RDD

其中hello.txt

回到顶部

二、RDD的创建方式

2.1　通过读取文件生成的

由外部存储系统的数据集创建，包括本地的文件系统，还有所有Hadoop支持的数据集，比如HDFS、Cassandra、HBase等

scala>valfile=sc.textFile("/spark/hello.txt")

2.2　通过并行化的方式创建RDD

由一个已经存在的Scala集合创建。

scala>valarray=Array(1,2,3,4,5)array:Array[Int]=Array(1,2,3,4,5)scala>valrdd=sc.parallelize(array)rdd:org.apache.spark.rdd.RDD[Int]=ParallelCollectionRDD[27]atparallelizeat<console>:26scala>

2.3　其他方式

读取数据库等等其他的操作。也可以生成RDD。

RDD可以通过其他的RDD转换而来的。

回到顶部

三、RDD编程API

Spark支持两个类型（算子）操作：Transformation和Action

3.1　Transformation

主要做的是就是将一个已有的RDD生成另外一个RDD。Transformation具有lazy特性(延迟加载)。Transformation算子的代码不会真正被执行。只有当我们的程序里面遇到一个action算子的时候，代码才会真正的被执行。这种设计让Spark更加有效率地运行。

常用的Transformation：

3.2　Action

触发代码的运行，我们一段spark代码里面至少需要有一个action操作。

常用的Action:

3.3　Spark WordCount代码编写

使用maven进行项目构建

（1）使用scala进行编写

查看官方网站，需要导入2个依赖包

详细代码

SparkWordCountWithScala.scala

importorg.apache.spark.rdd.RDDimportorg.apache.spark.{SparkConf,SparkContext}objectSparkWordCountWithScala{defmain(args:Array[String]):Unit={valconf=newSparkConf()/***如果这个参数不设置，默认认为你运行的是集群模式*如果设置成local代表运行的是local模式*/conf.setMaster("local")//设置任务名conf.setAppName("WordCount")//创建SparkCore的程序入口valsc=newSparkContext(conf)//读取文件生成RDDvalfile:RDD[String]=sc.textFile("E:\\hello.txt")//把每一行数据按照，分割valword:RDD[String]=file.flatMap(_.split(","))//让每一个单词都出现一次valwordOne:RDD[(String,Int)]=word.map((_,1))//单词计数valwordCount:RDD[(String,Int)]=wordOne.reduceByKey(_+_)//按照单词出现的次数降序排序valsortRdd:RDD[(String,Int)]=wordCount.sortBy(tuple=>tuple._2,false)//将最终的结果进行保存sortRdd.saveAsTextFile("E:\\result")sc.stop()}

运行结果

（2）使用java jdk7进行编写

SparkWordCountWithJava7.java

importorg.apache.spark.SparkConf;importorg.apache.spark.api.java.JavaPairRDD;importorg.apache.spark.api.java.JavaRDD;importorg.apache.spark.api.java.JavaSparkContext;importorg.apache.spark.api.java.function.FlatMapFunction;importorg.apache.spark.api.java.function.Function2;importorg.apache.spark.api.java.function.PairFunction;importscala.Tuple2;importjava.util.Arrays;importjava.util.Iterator;publicclassSparkWordCountWithJava7{publicstaticvoidmain(String[]args){SparkConfconf=newSparkConf();conf.setMaster("local");conf.setAppName("WordCount");JavaSparkContextsc=newJavaSparkContext(conf);JavaRDD<String>fileRdd=sc.textFile("E:\\hello.txt");JavaRDD<String>wordRDD=fileRdd.flatMap(newFlatMapFunction<String,String>(){@OverridepublicIterator<String>call(Stringline)throwsException{returnArrays.asList(line.split(",")).iterator();}});JavaPairRDD<String,Integer>wordOneRDD=wordRDD.mapToPair(newPairFunction<String,String,Integer>(){@OverridepublicTuple2<String,Integer>call(Stringword)throwsException{returnnewTuple2<>(word,1);}});JavaPairRDD<String,Integer>wordCountRDD=wordOneRDD.reduceByKey(newFunction2<Integer,Integer,Integer>(){@OverridepublicIntegercall(Integeri1,Integeri2)throwsException{returni1+i2;}});JavaPairRDD<Integer,String>count2WordRDD=wordCountRDD.mapToPair(newPairFunction<Tuple2<String,Integer>,Integer,String>(){@OverridepublicTuple2<Integer,String>call(Tuple2<String,Integer>tuple)throwsException{returnnewTuple2<>(tuple._2,tuple._1);}});JavaPairRDD<Integer,String>sortRDD=count2WordRDD.sortByKey(false);JavaPairRDD<String,Integer>resultRDD=sortRDD.mapToPair(newPairFunction<Tuple2<Integer,String>,String,Integer>(){@OverridepublicTuple2<String,Integer>call(Tuple2<Integer,String>tuple)throwsException{returnnewTuple2<>(tuple._2,tuple._1);}});resultRDD.saveAsTextFile("E:\\result7");}}

（3）使用java jdk8进行编写

lambda表达式

SparkWordCountWithJava8.java

importorg.apache.spark.SparkConf;importorg.apache.spark.api.java.JavaPairRDD;importorg.apache.spark.api.java.JavaRDD;importorg.apache.spark.api.java.JavaSparkContext;importscala.Tuple2;importjava.util.Arrays;publicclassSparkWordCountWithJava8{publicstaticvoidmain(String[]args){SparkConfconf=newSparkConf();conf.setAppName("WortCount");conf.setMaster("local");JavaSparkContextsc=newJavaSparkContext(conf);JavaRDD<String>fileRDD=sc.textFile("E:\\hello.txt");JavaRDD<String>wordRdd=fileRDD.flatMap(line->Arrays.asList(line.split(",")).iterator());JavaPairRDD<String,Integer>wordOneRDD=wordRdd.mapToPair(word->newTuple2<>(word,1));JavaPairRDD<String,Integer>wordCountRDD=wordOneRDD.reduceByKey((x,y)->x+y);JavaPairRDD<Integer,String>count2WordRDD=wordCountRDD.mapToPair(tuple->newTuple2<>(tuple._2,tuple._1));JavaPairRDD<Integer,String>sortRDD=count2WordRDD.sortByKey(false);JavaPairRDD<String,Integer>resultRDD=sortRDD.mapToPair(tuple->newTuple2<>(tuple._2,tuple._1));resultRDD.saveAsTextFile("E:\\result8");}

3.4　WordCount执行过程图

回到顶部

四、RDD的宽依赖和窄依赖

4.1　RDD依赖关系的本质内幕

由于RDD是粗粒度的操作数据集，每个Transformation操作都会生成一个新的RDD，所以RDD之间就会形成类似流水线的前后依赖关系；RDD和它依赖的父RDD（s）的关系有两种不同的类型，即窄依赖（narrow dependency）和宽依赖（wide dependency）。如图所示显示了RDD之间的依赖关系。

从图中可知：

窄依赖：是指每个父RDD的一个Partition最多被子RDD的一个Partition所使用，例如map、filter、union等操作都会产生窄依赖；（独生子女）

宽依赖：是指一个父RDD的Partition会被多个子RDD的Partition所使用，例如groupByKey、reduceByKey、sortByKey等操作都会产生宽依赖；（超生）

需要特别说明的是对join操作有两种情况：

（1）图中左半部分join：如果两个RDD在进行join操作时，一个RDD的partition仅仅和另一个RDD中已知个数的Partition进行join，那么这种类型的join操作就是窄依赖，例如图1中左半部分的join操作(join with inputs co-partitioned)；

（2）图中右半部分join：其它情况的join操作就是宽依赖,例如图1中右半部分的join操作(join with inputs not co-partitioned)，由于是需要父RDD的所有partition进行join的转换，这就涉及到了shuffle，因此这种类型的join操作也是宽依赖。

总结：

在这里我们是从父RDD的partition被使用的个数来定义窄依赖和宽依赖，因此可以用一句话概括下：如果父RDD的一个Partition被子RDD的一个Partition所使用就是窄依赖，否则的话就是宽依赖。因为是确定的partition数量的依赖关系，所以RDD之间的依赖关系就是窄依赖；由此我们可以得出一个推论：即窄依赖不仅包含一对一的窄依赖，还包含一对固定个数的窄依赖。

一对固定个数的窄依赖的理解：即子RDD的partition对父RDD依赖的Partition的数量不会随着RDD数据规模的改变而改变；换句话说，无论是有100T的数据量还是1P的数据量，在窄依赖中，子RDD所依赖的父RDD的partition的个数是确定的，而宽依赖是shuffle级别的，数据量越大，那么子RDD所依赖的父RDD的个数就越多，从而子RDD所依赖的父RDD的partition的个数也会变得越来越多。

4.2　依赖关系下的数据流视图

在spark中，会根据RDD之间的依赖关系将DAG图（有向无环图）划分为不同的阶段，对于窄依赖，由于partition依赖关系的确定性，partition的转换处理就可以在同一个线程里完成，窄依赖就被spark划分到同一个stage中，而对于宽依赖，只能等父RDD shuffle处理完成后，下一个stage才能开始接下来的计算。

因此spark划分stage的整体思路是：从后往前推，遇到宽依赖就断开，划分为一个stage；遇到窄依赖就将这个RDD加入该stage中。因此在图2中RDD C,RDD D,RDD E,RDDF被构建在一个stage中,RDD A被构建在一个单独的Stage中,而RDD B和RDD G又被构建在同一个stage中。

在spark中，Task的类型分为2种：ShuffleMapTask和ResultTask；

简单来说，DAG的最后一个阶段会为每个结果的partition生成一个ResultTask，即每个Stage里面的Task的数量是由该Stage中最后一个RDD的Partition的数量所决定的！而其余所有阶段都会生成ShuffleMapTask；之所以称之为ShuffleMapTask是因为它需要将自己的计算结果通过shuffle到下一个stage中；也就是说上图中的stage1和stage2相当于mapreduce中的Mapper,而ResultTask所代表的stage3就相当于mapreduce中的reducer。

在之前动手操作了一个wordcount程序，因此可知，Hadoop中MapReduce操作中的Mapper和Reducer在spark中的基本等量算子是map和reduceByKey;不过区别在于：Hadoop中的MapReduce天生就是排序的；而reduceByKey只是根据Key进行reduce，但spark除了这两个算子还有其他的算子；因此从这个意义上来说，Spark比Hadoop的计算算子更为丰富。

到此，关于“ Spark RDD怎么创建”的学习就结束了，希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习，快去试试吧！若想继续学习更多相关知识，请继续关注恰卡编程网网站，小编会继续努力为大家带来更多实用的文章！