Spark中RDD的设计思想

Posted 2023-02-15 暴雨骄阳

在大数据处理领域，开源项目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有一个简单的认识。