Spark中RDD的设计思想

Posted 暴雨骄阳

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Spark中RDD的设计思想相关的知识,希望对你有一定的参考价值。

在大数据处理领域,开源项目Hadoop曾经掀起过一场轩然大波,但随着使用者的逐渐增多,也逐渐出现一些技术问题,而后,出现了解决计算过程中JobTracker热点问题的YARN,也出现了解决MapReduce数据处理过程中IO消耗过大的Spark.

Spark 通过一种新的容错方式,达到减少磁盘和网络IO的开销,为了达到这个目的,他们设计了一种新的抽象数据结构RDD(Resilient Distributed Datasets — 弹性分布式数据集).用Spark的原话描述, RDD是可并行处理的只读数据分区后的集合,可以从HDFS上读取得到,也可以从其它的RDD得到,只要对一个RDD进行Transformation(后面会提到)操作,就会得到一个新的RDD,RDD中包含了从父RDD派生所必需的信息,从而不需要检查点操作就可以重构丢失的数据分区,通过lineage(血统) 可以不通过把中间结果保存在HDFS 来达到容错的目的,如果运行过程中节点出错,只需要重新结算即可恢复。

Spark 程序和普通的Scala 程序看上去没有什么区别,Spark提供了一种新的数据类型RDD,RDD是一个Scala类,这个类上定义了一系列集合的操作,如map, filter, join 等, 与Scala 集合类不同的是, RDD的操作不是在本地执行,而是通过调度器提交给Mesos, 然后分发到网络中的节点上运行的。Spark 追求的目标就是,编写分布式程序像单机程序那么简单,但它的数据和运行模式有很大不同,这也需要使用者具备很强的系统控制能力和分布式系统知识。

在scala 中的集合操作

val arrays = Seq("Error1", null,"Error2","Error3")
val count = arrays.filter(_ != null).count(_.contains("Error"))

在spark中做一个count

val file = spark.textFile(”hdfs:...”)  
val count = file.filter(line => line.contains("Error")).count()

RDD是Spark最基本的概念,是对分布式内存的抽象使用,实现了以操作本地集合来操作分布式集群上的数据集合的抽象实现,通常缓存在内存中,并且每次对数据集合操作的结果都存在内存中,下一个操作可以直接使用内存中的数据做为输入,这样可以在很大程度上缓解,hadoop的MapReduce框架由于shuffle
操作引起的大量磁盘IO.

RDD的操作类型可以分为两种:Transformation(T) 和 Action(A),从表面上看 T 输入时RDD 类型,输出也是RDD类型,而A输入类型是RDD,输出是一个值或者一个集合。从作业的角度来讲,T输入是一个中间过程,A输出是一个结果,从代码实现的角度看,T操作直涉及到元数据的操作,在本地执行,A操作会真正出发真实数据的读取与计算,在集群中执行,通过对RDD的操作完成了整个Spark程序。

如何创建RDD

  1. 从hdfs文件系统创建RDD

    如代码 val file = spark.textFile(”hdfs:…”) 这里的file 就是由外部HDFS文件创建出来的一个RDD

    查看textFile的定义,可以看到返回类型为RDD[String]
    def textFile(
    path: String,
    minPartitions: Int = defaultMinPartitions): RDD[String]

  2. 通过父RDD转换得到新的RDD

    如代码 file.filter(line => line.contains(“Error”)) 这里的filter 就是一个Transformation ,得到一个新的RDD

    查看filter的定义,可以看到返回类型为RDD
    def filter(f: T => Boolean): RDD[T]

RDD的内部属性
RDD 是一个抽象类,定义了5个关键的属性,每一个派生出的实体类都具备这五个基本的属性, Spark 也通过这些属性实现了对RDD的管理
1. 分区列表 — Partitions
2. 分区计算函数 — compute()
3. 父RDD的依赖关系 — Dependencies
4. Key-Value Pair 数据类型的分区器, 分区策略和分区数
5. 每个数据分区的地址列表

以HadoopRDD 为例
如例子中所示,方法textFile 通过HDFS创建了一个RDD[String] , 再进一步追踪源码,发现textFIle方法体内部调用了RDD的map方法,生成了一个新的RDD - MapPartitionsRDD 这个类包含了分区和具体的操作类,Hadoop 默认会以Hadoop的分片策略来创建分区,Hadoop的node信息记录在RDD中,以便在Action操作可以找到数据存储的真实地址。

def textFile(
      path: String,
      minPartitions: Int = defaultMinPartitions): RDD[String] = withScope 
    assertNotStopped()
    hadoopFile(path, classOf[TextInputFormat], classOf[LongWritable], classOf[Text],
      minPartitions).map(pair => pair._2.toString)
  

 def map[U: ClassTag](f: T => U): RDD[U] = withScope 
    val cleanF = sc.clean(f)
    new MapPartitionsRDD[U, T](this, (context, pid, iter) => iter.map(cleanF))
  

例子中的filter方法,是通过条件过滤数据,和上面的map方法处理逻辑类似,同样会生成以个新的MapPartitionsRDD

def filter(f: T => Boolean): RDD[T] = withScope 
    val cleanF = sc.clean(f)
    new MapPartitionsRDD[T, T](
      this,
      (context, pid, iter) => iter.filter(cleanF),
      preservesPartitioning = true)
  

上面的代码都是在做RDD转换, 这些过程中生成的类,会默认保存在内存中,当然内存不够时,也可以持久化到磁盘上,到现在为止,我们的程序运行,还没有向Hadoop提交任何job。
从count方法体可以看到,在做count时,Spark 才会真正向hadoop提交job

def count(): Long = sc.runJob(this, Utils.getIteratorSize _).sum

通过对Spark RDD 初步的介绍,希望能让读者对spark有一个简单的认识。

以上是关于Spark中RDD的设计思想的主要内容,如果未能解决你的问题,请参考以下文章

Spark学习——RDD的设计与运行原理

Spark:RDD的设计与运行原理

Spark之RDD算子-转换算子

Spark RDD:如何共享数据以进行并行操作

spark之RDD

3.4 Spark RDD运行架构