4. 第三章 RDD编程

Spark 中的RDD 就是一个不可变的分布式对象集合。每个RDD 都被分为多个分区，这些分区运行在集群中的不同节点上。

总的来说，每个Spark 程序或shell 会话都按如下方式工作：

(1) 从外部数据创建出输入RDD。
(2) 使用诸如filter() 这样的转化操作对RDD 进行转化，以定义新的RDD。
(3) 告诉Spark 对需要被重用的中间结果RDD 执行persist() 操作。
(4) 使用行动操作（例如count() 和first() 等）来触发一次并行计算，Spark 会对计算进行优化后再执行。
注：cache() 与使用默认存储级别调用persist() 是一样的。

1. 创建RDD

Spark 提供了两种创建RDD 的方式：读取外部数据集，以及在驱动器程序中对一个集合进行并行化。

1.1 使用parallelize()方法创建RDD

创建RDD 最简单的方式就是把程序中一个已有的集合传给SparkContext 的parallelize()方法.

//Scala中的parallelize()方法
val lines = sc.parallelize(List("pandas","I love China"))

//Java中的parallelize()方法
JavaRDD<String> lines = sc.parallelize(Arrays.asList("pandas", "i like pandas"));

1.2 从外部存储中读取数据来创建RDD

//Scala中的textFile()方法
val lines = sc.textFile("/path/to/README.md")

//Java中的textFile()方法
JavaRDD<String> lines = sc.textFile("/path/to/README.md");

2. RDD操作

2.1 转化操作

RDD的转化操作是返回新RDD的操作。转化出来的RDD是惰性求值的，只有在行动操作中用到这些RDD时才会被计算。

转化操作filter()和union()：

//用Scala实现filter()和union()转化操作
val inputRDD = sc.textFile("log.txt")
val errorsRDD = inputRDD.filter(line => line.contains("error"))
val warningsRDD = inputRDD.filter(line => line.contains("warning"))
badlinesRDD = errorsRDD.union(warningsRDD)

//用Java实现filter()和union()转化操作
JavaRDD<String> inputRDD = sc.textFile("log.txt");
JavaRDD<String> errorsRDD = inputRDD.filter(line -> line.contains("error"));
JavaRDD<String> warningsRDD = inputRDD.filter(line -> line.contains("warning"));
badlinesRDD = errorsRDD.union(warningsRDD)

日志分析过程中创建出的RDD谱系图

2.2 行动操作

行动操作是第二种类型的RDD 操作，它们会把最终求得的结果返回到驱动器程序，或者写入外部存储系统中。由于行动操作需要生成实际的输出，它们会强制执行那些求值必须用到的RDD 的转化操作。

行动操作count()：

//在Scala中使用行动操作对错误进行计数
println("Input had " + badLinesRDD.count() + " concerning lines")
println("Here are 10 examples:")
badLinesRDD.take(10).foreach(println)

//在Java中使用行动操作对错误进行计数
System.out.println("Input had " + badLinesRDD.count() + " concerning lines")
System.out.println("Here are 10 examples:")
for (String line: badLinesRDD.take(10)) {
  System.out.println(line);
}

2.3 惰性求值

RDD 的转化操作都是惰性求值的。惰性求值意味着当我们对RDD 调用转化操作（例如调用map()）时，操作不会立即执行。
为什么惰性求值：Spark 使用惰性求值，这样就可以把一些操作合并到一起来减少计算数据的步骤。

2.4 基本RDD操作

2.4.1. 两个最常用的转化操作是map()和filter()

从输入RDD映射与筛选得到的RDD

//Scala 计算RDD中各值的平方
val input = sc.parallelize(List(1,2,3,4))
val result = input.map(x => x*x)
println(result.collect().mkString(","))

//Java 计算RDD中各值的平方
JavaRDD<Integer> rdd = sc.parallelize(Arrays.asList(1, 2, 3, 4));
JavaRDD<Integer> result = rdd.map(x -> x*x);
System.out.println(StringUtils.join(result.collect(), ","));

2.4.2. flatMap和map的区别

RDD的flatMap和map的区别

//Scala中的flatMap()将行数据切分为单词
val lines = sc.parallelize(List("hello world", "hi"))
val words = lines.flatMap(line => line.split(" "))
words.first() // 返回"hello"

//Java 中的flatMap() 将行数据切分为单词
JavaRDD<String> lines = sc.parallelize(Arrays.asList("hello world", "hi"));
JavaRDD<String> words = lines.flatMap(line -> line.split(" "));
words.first(); // 返回"hello"

2.4.3. 伪集合操作
尽管RDD 本身不是严格意义上的集合，但它也支持许多数学上的集合操作，比如合并和相交操作。
注意：这些操作都要求操作的RDD是相同数据类型的。

RDD.distinct()
我们的RDD中最常缺失的集合属性是元素的唯一性，因为常常有重复的元素。如果只要唯一的元素，我们可以使用RDD.distinct() 转化操作来生成一个只包含不同元素的新RDD。
注意：distinct()操作的开销很大，因为它需要将所有数据通过网络进行混洗（shuffle），以确保每个元素都只有一份。

RDD.union(other)
union(other)，它会返回一个包含两个RDD中所有元素的RDD。
注意：与数学中的union()操作不同的是，如果输入的RDD中有重复数据，Spark的union()操作也会包含这些重复数据，我们可以使用distinct()方法来实现数学中的union()效果

RDD.intersection(other)
intersection(other) 方法，只返回两个RDD 中都有的元素。intersection()
在运行时也会去掉所有重复的元素（单个RDD 内的重复元素也会一起移除）
性能差，开销大，需要shuffle

RDD.subtract(other)
subtract(other)函数接收另一个RDD作为参数，返回一个由只存在于第一个RDD中而不存在于第二个RDD中的所有元素组成的RDD。
和intersection()一样，它也需要数据混洗。

一些简单的集合操作

笛卡尔积 RDD.cartesian(other)
cartesian(other) 转化操作会返回所有可能的(a, b) 对，其中a 是源RDD 中的元素，而b 则来自另一个RDD。
求大规模RDD的笛卡尔积开销巨大。

两个RDD的笛卡尔积

2.4.4 常见的行动操作

reduce操作：它接收一个函数作为参数，这个函数要操作两个RDD 的元素类型的数据并返回一个同样类型的新元素，常用于计算累加以及聚合操作。

//Scala中的reduce
val rdd = sc.parallelize(List(1,2,3,4,5))
val sum = rdd.reduce((x,y) => x+y) //sum = 15

//Java中的reduce
JavaRDD<Integer> rdd = sc.parallelize(Arrays.asList(1,2,3,4,5));
Integer sum = rdd.reduce((x,y) -> x+y);

fold操作
def fold(zeroValue: T)(op: (T, T) => T): T ,算子其实就是先对rdd分区的每一个分区进行使用op函数，在调用op函数过程中将zeroValue参与计算，最后在对每一个分区的结果调用op函数，同理此处zeroValue再次参与计算！。

//Scala
println(sc.parallelize(List(1, 2, 3, 4, 5, 6, 7, 8), 1).fold(10)((x,y) => x+y)) //56
println(sc.parallelize(List(1, 2, 3, 4, 5, 6, 7, 8), 2).fold(10)((x,y) => x+y)) //66

第一个就是一个分区，元素为：1,2,3,4,5,6,7,8 ，计算过程是现在该分区进行add求和(zeroValue参与计算)，所以是1+2+3+4+5+6+7+8+10完成分区的计算，结果为46，最后在合并所有分区结果，由于只有一个分区，所以计算过程为46+10=56！
第二个两个分区：1,2,3,4和5,6,7,8两个分区，对分区分别进行计算为1+2+3+4+10=20，5+6+7+8+10=36，最后合并分区结果为20+36+10=66，

aggregate操作
aggregate(zeroValue:U)(seqOp:(U,T) => U,comOp(U,U) => U):
zeroValue是一个初始值，seqOp对每一个分区操作的函数，combOp对每一个分区结果进行操作的函数。zeroValue在seqOp的每一次调用中都会参与函数计算，最后也参与combOp函数的计算。

//Scala 
//seqOp和combOp的第一个参数x来自于zeroValue:U
val rdd = sc.parallelize(List(1, 2, 3, 3), 2)  // 第二个参数2表示分区数
def seqOp(x: (Int,Int), y: Int): (Int, Int) = {
    println("sqpOP x:" + x + "\t" + y)
    (x._1 + y, x._2 + 1)
  } 
  
def combOp(x: (Int,Int), y: (Int, Int)): (Int, Int) = {
    println("combOp x:" + x + "\t y: " + y)
    (x._1 + y._1, x._2 + y._2)
  }

println(rdd.aggregate((3, 4))(seqOp, combOp))   // (18, 16)
// 使用函数式编程简写
rdd.aggregate((3, 4))(
     (x, y) => (x._1 + y, x._2 + 1),
     (x, y) => (x._1 + y._1, x._2 + y._2))

//Java
JavaRDD<Integer> rdd = sc.parallelize(Array.asList(1,2,3,3));  
//第二和第三个参数为函数的匿名实现（lambda形式）  
Tuple2<Double, Integer> result = rdd.aggregate(new Tuple2<Integer, Integer>(0, 0),   
                (x,y) -> new Tuple2<Integer, Integer>(x._1 + y, x._2 + 1),  
                (x,y) -> new Tuple2<Integer, Integer>(x._1 + y._1, x._2 + y._2)); 

首先是对序列生成两个分区的RDD，对每一个分区RDD求和和计数，在计算(1, 2) 时候，x._1 = 3 参与求和，即1+2+3= 6，x._2 = 4参与计数，即4+1+1= 6；对于(3, 3)求和和计数时zeroValue (3, 4)也参与计算，即求和为3+3+3= 9，计数为4+1+1= 6。最后调用combOp函数时候zeroValue (3, 4)再一次参与计算所以最后结果为(6+9+3= 18, 6+6+4= 16)。
注意：此处分区的划分是Spark按照List的顺序均分为“指定分区数”个分区

常用操作总结

RDD基本的转化操作 RDD基本的行动操作

3. 持久化（缓存）

Spark RDD是惰性求值的，所以同一个RDD多次调用行动操作，Spark每次都会重算RDD以及它的所有依赖。
这在迭代算法中消耗格外大，因为迭代算法常常会多次使用同一组数据。

//Scala中的两次执行行动操作
val result = rdd.map(x => x*x)
println(result.count())
println(result.collect().mkString(","))

为了避免多次计算同一个RDD，可以让Spark对数据进行持久化。当我们让Spark持久化存储一个RDD时，计算出RDD的节点会分别保存它们所求出的分区数据。如果一个有持久化数据的节点发生故障，Spark会在需要用到缓存的数据时重算丢失的数据分区。如果希望节点故障的情况不会拖累我们的执行速度，也可以把数据备份到多个节点上。

//Scala中使用persist()进行RDD的缓存
val result = rdd.map(x => x*x)
result.persist(StorageLevel.DISK_ONLY)
println(result.count())
println(result.collect().mkString(","))
result.unpersist() // 手动将持久化的RDD从缓存中移除

缓存级别

需要注意的几点：
1）RDD的cache()方法其实调用的就是persist方法，缓存策略默认为MEMORY_ONLY；
2）可以通过persist方法手工设定StorageLevel来满足工程需要的存储级别；
3）cache或者persist并不是action；
4）可以调用rdd.unpersist()手动清除缓存