Spark基础【RDD持久化、分区器、文件保存读取】

一 RDD持久化

1 RDD Cache缓存

（1）为什么使用Cache

RDD通过Cache或者Persist方法将前面的计算结果缓存，默认情况下会把数据以缓存在JVM的堆内存中。但是并不是这两个方法被调用时立即缓存，而是触发后面的action算子时，该RDD将会被缓存在计算节点的内存中，并供后面重用

如现有两个需求：

• File – textFile – flatMap – map – reduceByKey
• File – textFile – flatMap – map – group

因为前四个阶段相同，能否直接在第一个需求的map后面进行更改？

def main(args: Array[String]): Unit = {
   

  val conf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
  val sc = new SparkContext(conf)

  val lines: RDD[String] = sc.textFile("data/word.txt")
  val words: RDD[String] = lines.flatMap(_.split(" "))
  val wordToOne: RDD[(String, Int)] = words.map(
    t => {
   
      println("**************")
      (t,1)
    }
  )
  val wordCount: RDD[(String, Int)] = wordToOne.reduceByKey(_ + _)
  wordCount.collect().foreach(println)

  println("----------------------")
  val rdd2: RDD[(String, Iterable[(String, Int)])] = wordToOne.groupBy(_._1)
  rdd2.collect().foreach(println)

  sc.stop()
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

执行结果

**************
**************
**************
**************
**************
**************
(scala,1)
(hello,3)
(world,1)
(spark,1)
----------------------
**************
**************
**************
**************
**************
**************
(scala,CompactBuffer((scala,1)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17