RDD的创建与操作

落日映苍穹つ 2021-09-14 03:56 344阅读 0赞

# 一、RDD的创建 #

进行Spark核心编程时，首先要做的第一件事，就是创建一个初始的RDD。该RDD中，通常就代表和包含了Spark应用程序的输入源数据。然后在创建了初始的RDD之后，才可以通过Spark Core提供的transformation算子，对该RDD进行转换，来获取其他的RDD。

Spark Core提供了三种创建RDD的方式，包括：使用程序中的集合创建RDD；使用本地文件创建RDD；使用HDFS文件创建RDD。

如果要通过并行化集合来创建RDD，需要针对程序中的集合，调用SparkContext的parallelize()方法。Spark会将集合中的数据拷贝到集群上去，形成一个分布式的数据集合，也就是一个RDD。相当于是，集合中的部分数据会到一个节点上，而另一部分数据会到其他节点上。然后就可以用并行的方式来操作这个分布式数据集合，即RDD。

调用parallelize()时，有一个重要的参数可以指定，就是要将集合切分成多少个partition。Spark会为每一个partition运行一个task来进行处理。Spark官方的建议是，为集群中的每个CPU创建2~4个partition。Spark默认会根据集群的情况来设置partition的数量。但是也可以在调用parallelize()方法时，传入第二个参数，来设置RDD的partition数量。比如parallelize(arr, 10)。

Spark是支持使用任何Hadoop支持的存储系统上的文件创建RDD的，比如说HDFS、Cassandra、HBase以及本地文件。通过调用SparkContext的textFile()方法，可以针对本地文件或HDFS文件创建RDD。

有几个事项是需要注意的：  
      1、如果是针对本地文件的话，如果是在windows上本地测试，windows上有一份文件即可；如果是在spark集群上针对linux本地文件，那么需要将文件拷贝到所有worker节点上。  
      2、Spark的textFile()方法支持针对目录、压缩文件以及通配符进行RDD创建。  
      3、Spark默认会为hdfs文件的每一个block创建一个partition，但是也可以通过textFile()的第二个参数手动设置分区数量，只能比block数量多，不能比block数量少。

Spark的textFile()除了可以针对上述几种普通的文件创建RDD之外，还有一些特列的方法来创建RDD：  
      1、SparkContext.wholeTextFiles()方法，可以针对一个目录中的大量小文件，返回<filename, fileContent>组成的pair，作为一个PairRDD，而不是普通的RDD。普通的textFile()返回的RDD中，每个元素就是文件中的一行文本。  
      2、SparkContext.sequenceFile\[K, V\]()方法，可以针对SequenceFile创建RDD，K和V泛型类型就是SequenceFile的key和value的类型。K和V要求必须是Hadoop的序列化类型，比如IntWritable、Text等。  
      3、SparkContext.hadoopRDD()方法，对于Hadoop的自定义输入类型，可以创建RDD。该方法接收JobConf、InputFormatClass、Key和Value的Class。  
      4、SparkContext.objectFile()方法，可以针对之前调用RDD.saveAsObjectFile()创建的对象序列化的文件，反序列化文件中的数据，并创建一个RDD。

# 二、RDD的操作 #

Spark支持两种RDD操作：transformation和action。transformation操作会针对已有的RDD创建一个新的RDD；而action则主要是对RDD进行最后的操作，比如遍历、reduce、保存到文件等，并可以返回结果给Driver程序。  
       例如，map就是一种transformation操作，它用于将已有RDD的每个元素传入一个自定义的函数，并获取一个新的元素，然后将所有的新元素组成一个新的RDD。而reduce就是一种action操作，它用于对RDD中的所有元素进行聚合操作，并获取一个最终的结果，然后返回给Driver程序。  
       transformation的特点就是lazy特性。lazy特性指的是，如果一个spark应用中只定义了transformation操作，那么即使你执行该应用，这些操作也不会执行。也就是说，transformation是不会触发spark程序的执行的，它们只是记录了对RDD所做的操作，但是不会自发的执行。只有当transformation之后，接着执行了一个action操作，那么所有的transformation才会执行。Spark通过这种lazy特性，来进行底层的spark应用执行的优化，避免产生过多中间结果。

例如：统计文件字数这个案例。  
        这里通过textFile()方法，针对外部文件创建了一个RDD，lines，但是实际上，程序执行到这里为止，sparktest.txt文件的数据是不会加载到内存中的。lines，只是代表了一个指向sparktest.txt文件的引用。  
val lines = sc.textFile("sparktest.txt")  
  
        这里对lines RDD进行了map算子，获取了一个转换后的lineLengths RDD。但是这里连数据都没有，当然也不会做任何操作。lineLengths RDD也只是一个概念上的东西而已。  
val lineLengths = lines.map(line => line.length)  
  
         这里执行了一个action操作，reduce。此时就会触发之前所有transformation操作的执行，Spark会将操作拆分成多个task到多个机器上并行执行，每个task会在本地执行map操作，并且进行本地的reduce聚合。最后会进行一个全局的reduce聚合，然后将结果返回给Driver程序。  
val totalLength = lineLengths.reduce(\_ + \_)  
  
         action操作执行，会触发一个spark job的运行，从而触发这个action之前所有的transformation的执行。这是action的特性。

Spark有些特殊的算子，也就是特殊的transformation操作。比如groupByKey、sortByKey、reduceByKey等，其实只是针对特殊的RDD的。即包含key-value对的RDD。而这种RDD中的元素，实际上是scala中的一种类型，即Tuple2，也就是包含两个值的Tuple。  
         在Scala中，需要手动导入Spark的相关隐式转换，import org.apache.spark.SparkContext.\_。然后，对应包含Tuple2的RDD，会自动隐式转换为PairRDDFunction，并提供reduceByKey等方法。

例如：  
val lines = sc.textFile("sparktest.txt")  
val linePairs = lines.map(line => (line, 1))  
val lineCounts = linePairs.reduceByKey(\_ + \_)  
lineCounts.foreach(lineCount => println(lineCount.\_1 + " appears " + llineCount.\_2 + " times."))

常用的transformation操作：

> > > > ![Center][]
> > > > 
> > > >   
> > > > 
> > > > 
> > > >   
> > > > 
> > > > 
> > > >   
> > > > 
> > > > 
> > > >   
> > > >

常用的action操作：

> > > > ![Center 1][]

> > > >   
> > > >

[Center]: /images/20210726/a4dae50e55394fcb9e0ca6bd893049c4.png
[Center 1]: /images/20210726/39562fde7faa4aa7b51bd8e9add1b242.png