RDD详解

RDD持久化/缓存

• 某些RDD的计算或转换可能会比较耗费时间，如果这些RDD后续还会频繁的被使用到，那么可以将这些RDD进行持久化/缓存

val rdd1 = sc.textFile("hdfs://node01:8020/words.txt")
val rdd2 = rdd1.flatMap(x=>x.split(" ")).map((_,1)).reduceByKey(_+_)
rdd2.cache //缓存/持久化
rdd2.sortBy(_._2,false).collect//触发action,会去读取HDFS的文件,rdd2会真正执行持久化
rdd2.sortBy(_._2,false).collect//触发action,会去读缓存中的数据,执行速度会比之前快,因为rdd2已经持久化到内存中了

• 创建了一个RDD，从HDFS上指定的路径"hdfs://node01:8020/words.txt"读取文本文件，每行文本都是RDD中的一个元素
• flatMap(x=>x.split(" "))：将每行文本拆分成单词，使用空格作为分隔符
• map((_,1))：对于每个单词，创建一个键值对，键是单词本身，值是1
• reduceByKey(+)：对每个键（单词）的值进行求和，这样每个单词的计数就计算出来了
• sortBy(_._2,false)：对RDD中的元素按照它们的值（即单词计数）进行排序，_._2指的是元组的第二个元素，false参数表示降序排序
• collect：这是一个行动操作，它会触发实际的计算，并将结果作为数组返回给驱动程序

persist方法和cache方法

• RDD通过persist或cache方法可以将前面的计算结果缓存，但是并不是这两个方法被调用时立即缓存，而是触发后面的action时，该RDD将会被缓存在计算节点的内存中，并供后面重用

存储级别

• 默认的存储级别都是仅在内存存储一份，Spark的存储级别还有好多种，存储级别在object StorageLevel中定义的

RDD容错机制Checkpoint

• Checkpoint的产生就是为了更加可靠的数据持久化，在Checkpoint的时候一般把数据放在在HDFS上，这就天然的借助了HDFS天生的高容错、高可靠来实现数据最大程度上的安全，实现了RDD的容错和高可用

SparkContext.setCheckpointDir("目录") //HDFS的目录

RDD.checkpoint

• 开发中如何保证数据的安全性性及读取效率：可以对频繁使用且重要的数据，先做缓存/持久化，再做checkpint操作

RDD的依赖关系

• RDD有两种依赖，分别为宽依赖(wide dependency/shuffle dependency)和窄依赖(narrow dependency)

窄依赖

• 父RDD的一个分区只会被子RDD的一个分区依赖
• 窄依赖的多个分区可以并行计算
• 窄依赖的一个分区的数据如果丢失只需要重新计算对应的分区的数据就可以了

宽依赖

• 父RDD的一个分区会被子RDD的多个分区依赖(涉及到shuffle)
• 划分Stage(阶段)的依据：对于宽依赖，必须等到上一阶段计算完成才能计算下一阶段