|
第一次分享的妥妥就是入门的干货，小伙伴们最好可以自己敲一敲代码，不然只看我的分享一点用都木有。但还是有很多小伙伴表示看不懂，没关系，慢慢来自己操作一遍有什么问题后台问我就可以了。 啥也不说了，就是干货，首先祭上今天的关键代码。今天主要跟大家讲讲Spark里面RDD的持久化机制。首先持久化机制有什么用呢？一个作用是保存到硬盘给其他小伙伴查看，另外一个作用是重用，我们都知道RDD是不可变的，所以当RDD有重用的时候，如果没有持久化，RDD都会乖乖地重新算。。 objectRDDPersist { defmain(args: Array[String]): Unit = { valconf: SparkConf =newSparkConf().setMaster('local[*]').setAppName('helloWorld') valsc: SparkContext =newSparkContext(conf); sc.setCheckpointDir('~/checkpoint'); varsomeList =List[String]('hello','world') //我们先假设这里有一百万的数据 valdefaultPartition =5; valwordsRDD: RDD[String] = sc.parallelize(someList,defaultPartition); valsomeThingSenceless = wordsRDD.map(x => x ',').map(x => ((Math.floor(Math.random()*100)) ,x)).filter(x => x._1 <50) someThingSenceless.count(); someThingSenceless.groupByKey().count(); wordsRDD.repartition(defaultPartition*2); wordsRDD.coalesce(defaultPartition ,false) wordsRDD.cache() wordsRDD.persist(StorageLevel.MEMORY_ONLY); /* StorageLevel.DISK_ONLY StorageLevel.DISK_ONLY_2 StorageLevel.MEMORY_ONLY StorageLevel.MEMORY_ONLY_2 StorageLevel.MEMORY_ONLY_SER StorageLevel.MEMORY_ONLY_SER_2 StorageLevel.MEMORY_AND_DISK StorageLevel.MEMORY_AND_DISK_2 StorageLevel.MEMORY_AND_DISK_SER StorageLevel.MEMORY_AND_DISK_SER_2 */ wordsRDD.saveAsTextFile('~/') wordsRDD.unpersist(); wordsRDD.checkpoint(); } 从前面的分享我们可以知道，Spark快的根本原因就是Spark的中间过程都存储在内存里面，而这一切的基础就是RDD的持久化，RDD默认是根据partition按分区持久化在内存里边的。啥意思呢，一份数据就好像一个钥匙串，而这钥匙串会分成好多根钥匙保存到集群机器上。 varsomeList =List[String]('hello','world') //我们先假设这里有一百万的数据 valdefaultPartition =5; valwordsRDD: RDD[String] = sc.parallelize(someList,defaultPartition); 就是初始化了一个有一百万数据的RDD，分布在5个分区里面。 五个分区好像有点少，想分配到更多的分区咋办？ wordsRDD.repartition(defaultPartition*2); wordsRDD.coalesce(defaultPartition ,false) coalesce是一个重新分区操作，第一个参数是目标的分区数，第二个参数是是否进行suffle，也就是混淆重新洗牌。如果不重新洗牌，那么就会按照分区数一对一分配到更多的分区上，比如一个分区拆分成两个分区这样，而不是把两个分区进行混淆重新洗牌然后平均分配到三个分区里边。怎么理解这个概念呢，看下面这个图，文字可以忽略，左边就是没有suffle的，右边是有的。 repartition是coalesce的简单实现，就是把第二个参数默认为true，就是repartition了。 valsomeThingSenceless = wordsRDD.map(x => x ',').map(x => ((Math.floor(Math.random()*100)) ,x)).filter(x => x._1 <50) someThingSenceless.count(); someThingSenceless.groupByKey().count(); 这里的操作都没什么意义，就是做一些复杂的操作而已。如果我们不对wordsRDD 进行持久化，那么一切的操作都会从源头开始，一步一步往下算，不会复用原始的数据。这明显不是我们想要的，如果我们不想这样，那要怎么办呢？ wordsRDD.cache() wordsRDD.persist(StorageLevel.MEMORY_ONLY); /* StorageLevel.DISK_ONLY StorageLevel.DISK_ONLY_2 StorageLevel.MEMORY_ONLY StorageLevel.MEMORY_ONLY_2 StorageLevel.MEMORY_ONLY_SER StorageLevel.MEMORY_ONLY_SER_2 StorageLevel.MEMORY_AND_DISK StorageLevel.MEMORY_AND_DISK_2 StorageLevel.MEMORY_AND_DISK_SER StorageLevel.MEMORY_AND_DISK_SER_2 */ 虽然说Spark的东西都尽量放到内存里边，但是对于非常非常巨大的数据集，Spark能拿它怎么办呢？总不能全丢进去然后丢弃一大批吧？所以就有了以下的策略选择了。 cache()方法很好理解，就是全都放在内存里面，也就是MEMORY_ONLY。persist方法的默认参数也是仅内存。MEMORY_AND_DISK的意思就是，内存能放得下就放内存，放不下就放硬盘。如何理解这几个参数的搭配呢？从三个要素来搭配就行了，存储位置、存储份数，是否序列化。三个的所有搭配都在上面的。 通常建议是使用StorageLevel.MEMORY_AND_DISK_SER。为什么使用这个呢？比较明显的原因，就是可以通过序列化压缩的方式，保证内存利用最大化。 wordsRDD.saveAsTextFile('~/') 细心的观众可能会发现，这又是什么玩意。这就是把目前的数据集直接保存到Driver硬盘上的一个API而已。 我们都知道RDD是有parent的，也就是RDD是一个有向无环图，那么如果调用链太长太长了，那么最后的一个操作引起机器失联，会导致整个计算从头开始，这样的开销非常非常大，特别是在深度非常深的迭代操作中，那怎么解决这个问题呢？ sc.setCheckpointDir('~/checkpoint'); wordsRDD.checkpoint(); 答案就是使用checkpoint，这个是啥意思呢，就是把RDD保存到磁盘或者HDFS上，直接把当前RDD当成顶级RDD，也就没有所谓RDD的依赖链了，可以缓解这个多次迭代导致的问题。但是世界上哪有这么好的事，checkpoint的代价就是RDD保存在硬盘上，效率总是要降低的。 数据没用了咋办？删掉呗。无论是什么方式，都可以使用unpersist这个API进行删除所有的持久化。 wordsRDD.unpersist(); 好了今天的分享就到这里，RDD的管理只是Spark的内存管理的一部分，优先理解这部分内容，对开发Spark应用优化有非常重要的作用，就酱。 如果分享的话，我会很开心的。 |
|
|
