RDD算子之后

2021-12-18  本文已影响0人  初夏时的猫

RDD序列化

闭包检查

从计算的角度,**算子以外的代码都是在Driver端执行,算子里面的代码都是在Executor端执行。**那么在scala函数式编程中,就会导致算子内经常会用到算子外的数据,这样就形成了闭包的效果,如果使用的算子外的数据无法序列化,就意味着无法传值给Executor端执行,从而发生错误,所以需要在执行任务计算前,检测闭包内对象是否可以进行序列化,这个操作我们称之为闭包检测。

Kryo序列化框架

Java的序列化能够序列化任何的类。单比较重(字节多),序列化后,对象的提交也比较大。Spark2.0开始支持Kryo序列化机制。Kryo速度是Serializable的10倍。当RDD子shuffle数据时,简单数据类型、数组、字符串类型已经在Spark内部使用Kryo来序列化。

object SerTest {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setMaster("local[*]")
      .setAppName("serTest")
      .set("spark.serializer","org.apache.spark.serializer.KryoSerializer")
    val sc = new SparkContext(conf)
    val rdd1 = sc.makeRDD(List("hello spark","hello scala"))
    val person:Person = new Person("hel")
    val rdd2 = person.getMatchedRDD1(rdd1)
    rdd2.collect().foreach(println)
  }
}
class Person(str:String) extends Serializable {
    def isMatch(s:String)={
      s.contains(str)
    }

    def getMatchedRDD1(rdd:RDD[String]): RDD[String] ={
      rdd.filter(isMatch)
    }
}

RDD依赖关系

图片.png

1.血缘关系

  图中的rdd之间存在血缘关系
  rdd.toDebugString

2.依赖关系

  每个相邻rdd之间存在依赖关系
  rdd.dependencies
  
1)rdd窄依赖
class OneToOneDependency[T](rdd: RDD[T]) extends NarrowDependency[T](rdd)
2)rdd宽依赖
class ShuffleDependency[K: ClassTag, V: ClassTag, C: ClassTag](
@transient private val _rdd: RDD[_ <: Product2[K, V]],
val partitioner: Partitioner,
val serializer: Serializer = SparkEnv.get.serializer,
val keyOrdering: Option[Ordering[K]] = None,
val aggregator: Option[Aggregator[K, V, C]] = None,
val mapSideCombine: Boolean = false)
extends Dependency[Product2[K, V]]

rdd阶段划分&&rdd任务划分

url

rdd缓存和检查点

rdd.cache

rdd.persist(StorageLevel.缓存方式)

sc.checkpointDir(path)
rdd.checkpoint
注:以上操作要执行行动算子后才进行存储。因为只有行动算子被执行,driver端才会拿到executor端计算的数据。


图片.png

缓存和检查的区别

1) Cache 缓存只是将数据保存起来,不切断血缘依赖。 Checkpoint 检查点切断血缘依赖。

2) Cache 缓存的数据通常存储在磁盘、内存等地方,可靠性低。 Checkpoint 的数据通常存
储在 HDFS 等容错、高可用的文件系统,可靠性高。

3)建议对 checkpoint()的 RDD 使用 Cache 缓存,这样 checkpoint 的 job 只需从 Cache 缓存
中读取数据即可,否则需要再从头计算一次 RDD。

自定义分区(类似mr的分区)

1.继承Partitioner抽象类
2.实现其中方法

方法1:numPartitions:Int,返回分区数量即可

方法2:getPartition(key:Any): Boolean 写具体分区逻辑。

文件读写

1.text文件

2.sequence文件:Hadoop用来存储二进制形式的key-value而设计的一种平面文件(FlatFile)。

3.object对象文件:对象文件是将对象序列化后保存的文件,采用Java的序列化机制。

上一篇下一篇

猜你喜欢

热点阅读