专栏首页魏晓蕾的专栏弹性分布式数据集RDD及其操作

弹性分布式数据集RDD及其操作

1、RDD简介

RDD是Spark提供的核心抽象,全称为Resillient Distributed Dataset,即弹性分布式数据集。RDD在抽象上来说是一种元素集合,包含了数据。它是被分区的,分为多个分区,每个分区分布在集群中的不同节点上,从而让RDD中的数据可以被并行操作(分布式数据集)。RDD通常通过Hadoop上的文件,即HDFS文件或者Hive表,来进行创建;有时也可以通过应用程序中的集合来创建。RDD最重要的特性就是,提供了容错性,可以自动从节点失败中恢复过来。即如果某个节点上的RDD partition,因为节点故障,导致数据丢了,那么RDD会自动通过自己的数据来源重新计算该partition。这一切对使用者是透明的。RDD的数据默认情况下存放在内存中的,但是在内存资源不足时,Spark会自动将RDD数据写入磁盘(弹性)。 RDD分布图如下:

2、RDD的特性

(1)RDD的创建 进行Spark核心编程时,首先要做的第一件事,就是创建一个初始的RDD。该RDD中,通常就代表和包含了Spark应用程序的输入源数据。然后在创建了初始的RDD之后,才可以通过Spark Core提供的transformation算子,对该RDD进行转换,来获取其他的RDD。 Spark Core提供了三种创建RDD的方式,包括:使用程序中的集合创建RDD;使用本地文件创建RDD;使用HDFS文件创建RDD。 1)使用程序中的集合创建RDD,主要用于进行测试,可以在实际部署到集群运行之前,自己使用集合构造测试数据,来测试后面的spark应用的流程。 2)使用本地文件创建RDD,主要用于临时性地处理一些存储了大量数据的文件。 3)使用HDFS文件创建RDD,应该是最常用的生产环境处理方式,主要可以针对HDFS上存储的大数据,进行离线批处理操作。 1)并行化集合(内存中数据)创建RDD:基于序列化进行创建 如果要通过并行化集合来创建RDD,需要针对程序中的集合,调用SparkContext的parallelize()方法。Spark会将集合中的数据拷贝到集群上去,形成一个分布式的数据集合,也就是一个RDD。相当于是,集合中的部分数据会到一个节点上,而另一部分数据会到其他节点上。然后就可以用并行的方式来操作这个分布式数据集合,即RDD。 案例1:1到10累加求和

val arr = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
val rdd = sc.parallelize(arr)
val sum = rdd.reduce(_ + _)

scala> val seq = List(1,2,3,4,5,6,7)
seq: List[Int] = List(1, 2, 3, 4, 5, 6, 7)
scala> val rdd2 = sc.parallelize(seq)
rdd2: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[4] at parallelize at <console>:29

调用parallelize()时,有一个重要的参数可以指定,就是要将集合切分成多少个partition。Spark会为每一个partition运行一个task来进行处理。Spark官方的建议是,为集群中的每个CPU创建2~4个partition。Spark默认会根据集群的情况来设置partition的数量。但是也可以在调用parallelize()方法时,传入第二个参数,来设置RDD的partition数量。比如parallelize(arr, 10)。 2)使用本地文件和HDFS(外部数据)创建RDD park是支持使用任何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数量少。

案例2:文件字数统计

val rdd = sc.textFile("data.txt")
val wordCount = rdd.map(line => line.length).reduce(_ + _)

Spark的textFile()除了可以针对上述几种普通的文件创建RDD之外,还有一些特列的方法来创建RDD:

  1. SparkContext.wholeTextFiles()方法,可以针对一个目录中的大量小文件,返回<filename, fileContent>组成的pair,作为一个PairRDD,而不是普通的RDD。普通的textFile()返回的RDD中,每个元素就是文件中的一行文本。
  2. SparkContext.sequenceFileK, 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。

基于MapReduce的InputFormat进行创建sc.textFile():底层使用TextInputFormat读取数据形成RDD,此方法为使用旧API; sc.newAPIHadoopFile():底层使用TextInputFormat读取数据形成RDD,此方法为使用新API; sc.newAPIHadoopRDD():API指定使用哪个InputFormat读取数据。 (2)RDD的依赖 1)窄依赖 子RDD的每个分区的数据来自常数个父RDD分区,父RDD的每个分区的数据到子RDD的时候在一个分区中进行处理。常用方法:map、flatmap、filter、union、join(要求两个父RDD具有相同的partitioner,同时两个父rdd的分区数目和子rdd的分区数目一致)等。 子 RDD 的每个分区依赖于常数个父分区(即与数据规模无关)。 输入输出一对一的算子,且结果 RDD 的分区结构不变,主要是 map 、 flatMap。 输入输出一对一的算子,但结果 RDD 的分区结构发生了变化,如 union 、 coalesce(要求shuffle参数为false)。 从输入中选择部分元素的算子,如 filter 、 subtract 、 sample。 2)宽依赖 子RDD的每个分区的数据来自所有的父RDD分区,父RDD的每个分区的数据都有可能分配到所有的子RDD分区中。常用方法:xxxxByKey、join、repartition等。 子 RDD 的每个分区依赖于所有父 RDD 分区。 对单个 RDD 基于 key 进行重组和 reduce,如 groupByKey 、distinct 、 reduceByKey 。 对两个 RDD 基于 key 进行 join 和重组,如 join(分区数量进行改变)。

3、RDD的操作

(1)Transformation和Action 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所做的操作,但是不会自发的执行。在这些类型的API调用过程中,只会构建RDD的依赖,也称为构建RDD的执行逻辑图(DAG图)。只有当transformation之后,接着执行了一个action操作,那么所有的transformation才会执行。Action操作触发RDD的job执行提交操作,并将RDD对应的job提交到executor上执行。该类型的API调用的时候,会触发job的执行,并将job的具体执行过程提交到executor上执行,最终的执行结果要不输出到其它文件系统或者返回给driverSpark通过这种lazy特性,来进行底层的spark应用执行的优化,避免产生过多中间结果。 action操作执行,会触发一个spark job的运行,从而触发这个action之前所有的transformation的执行。这是action的特性。 案例3:统计文件字数 这里通过textFile()方法,针对外部文件创建了一个RDD,lines,但是实际上,程序执行到这里为止,spark.txt文件的数据是不会加载到内存中的。lines,只是代表了一个指向spark.txt文件的引用。

val lines = sc.textFile("spark.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(_ + _)

案例4:统计文件每行出现的次数 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("hello.txt")
val linePairs = lines.map(line => (line, 1))
val lineCounts = linePairs.reduceByKey(_ + _)
lineCounts.foreach(lineCount => println(lineCount._1 + " appears " + llineCount._2 + " times."))

(2)RDD Transformation有如下方法: map(),flatMap(),filter(),mapPartitions(),mapPartitionsWithIndex(),sample(),union(),intersection(),distinct(),groupByKey(),reduceByKey(),sortByKey(),join(),cogroup(),cartesion(),pipe(),coalesce(),repartition(),partitionBy() RDD Action有如下方法: reduce(),collect(),count(),first(),take(),takeSample(),saveToCassandra(),takeOrdered(),saveAsTextFile(),saveAsSequenceFile(),saveAsObjectFile(),countByKey(),foreach() (3)常用Transformation介绍

操作

介绍

map

将RDD中的每个元素传入自定义函数,获取一个新的元素,然后用新的元素组成新的RDD

filter

对RDD中每个元素进行判断,如果返回true则保留,返回false则剔除

flatMap

与map类似,但是对每个元素都可以返回一个或多个新元素

gropuByKey

根据key进行分组,每个key对应一个Iterable< value>

reduceByKey

对每个key对应的value进行reduce操作

sortByKey

对每个key对应的value进行排序操作

join

对两个包含<key,value>对的RDD进行join操作,每个key join上的pair,都会传入自定义函数进行处理

cogroup

同join,但是是每个key对应的Iterable< value>都会传入自定义函数进行处理

常用Action介绍

操作

介绍

reduce

将RDD中的所有元素进行聚合操作。第一个和第二个元素聚合,值与第三个元素聚合,值与第四个元素聚合,以此类推

collect

将RDD中所有元素获取到本地客户端

count

获取RDD元素总数

take(n)

获取RDD中前n个元素

saveAsTextFile

将RDD元素保存到文件中,对每个元素调用toString方法

countByKey

对每个key对应的值进行count计数

foreach

遍历RDD中的每个元素

(4)Transformation和Action原理剖析

4、RDD的持久化

Spark非常重要的一个功能特性就是可以将RDD持久化在内存中。当对RDD执行持久化操作时,每个节点都会将自己操作的RDD的partition持久化到内存中,并且在之后对该RDD的反复使用中,直接使用内存缓存的partition。这样的话,对于针对一个RDD反复执行多个操作的场景,就只要对RDD计算一次即可,后面直接使用该RDD,而不需要反复计算多次该RDD。 巧妙使用RDD持久化,甚至在某些场景下,可以将spark应用程序的性能提升10倍。对于迭代式算法和快速交互式应用来说,RDD持久化,是非常重要的。 要持久化一个RDD,只要调用其cache()或者persist()方法即可。在该RDD第一次被计算出来时,就会直接缓存在每个节点中。而且Spark的持久化机制还是自动容错的,如果持久化的RDD的任何partition丢失了,那么Spark会自动通过其源RDD,使用transformation操作重新计算该partition。 cache()和persist()的区别在于,cache()是persist()的一种简化方式,cache()的底层就是调用的persist()的无参版本,同时就是调用persist(MEMORY_ONLY),将数据持久化到内存中。如果需要从内存中清楚缓存,那么可以使用unpersist()方法。 Spark自己也会在shuffle操作时,进行数据的持久化,比如写入磁盘,主要是为了在节点失败时,避免需要重新计算整个过程。 rdd.cache():数据缓存到内存中。 rdd.persist(xxx):数据缓存到指定级别的存储系统中(内存\内存+磁盘\磁盘)。 rdd.unpersist():清除缓存数据。 不使用RDD持久化的问题的原理:

RDD持久化的工作原理:

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • 【OverFeat】Integrated Recognition, Localization and Detection using Convolutional Networks(2014) 全文翻译

    Pierre Sermanet, David Eigen, Xiang Zhang, Michael Mathieu, Rob Fergus, Yann LeC...

    魏晓蕾
  • Scala中的模式匹配、类型参数与隐式转换

    (1)概述 模式匹配是Scala中非常有特色,非常强大的一种功能。模式匹配,其实类似于Java中的swich case语法,即对一个值进行条件判断,然后针对不...

    魏晓蕾
  • 【Spark】Spark Streaming的程序运行原理及与Kafka的集成

    (1)Streaming Streaming:是一种数据传送技术,它把客户机收到的数据变成一个稳定连续的流,源源不断地送出,使用户听到的声音或看到的图象十分平...

    魏晓蕾
  • 从零到一spark进阶之路(一)

    RDD是spark特有的数据模型,谈到RDD就会提到什么弹性分布式数据集,什么有向无环图,本文暂时不去展开这些高深概念。

    学到老
  • Spark核心RDD、什么是RDD、RDD的属性、创建RDD、RDD的依赖以及缓存、

    1:什么是Spark的RDD??? RDD(Resilient Distributed Dataset)叫做分布式数据集,是Spark中最基本的数据抽象,它代...

    别先生
  • Spark Core快速入门系列(2) | Spark Core中编程模型的理解与RDD的创建

      在 Spark 中,RDD 被表示为对象,通过对象上的方法调用来对 RDD 进行转换。   经过一系列的transformations定义 RDD 之后,...

    不温卜火
  • Spark计算RDD介绍

    用户2196435
  • Spark Core——RDD何以替代Hadoop MapReduce?

    继续前期依次推文PySpark入门和SQL DataFrame简介的基础上,今日对Spark中最重要的一个概念——RDD进行介绍。虽然在Spark中,基于RDD...

    luanhz
  • spark RDD 结构最详解

    一种简单的解释RDD是横向多分区的(这个数据集包括许多接口),纵向当计算过程中内存不足可刷写到磁盘等外存上,可与外存进行灵活的数据交换。

    黑白格
  • 了解Spark中的RDD

    用户2196435

扫码关注云+社区

领取腾讯云代金券