Spark基础-RDD编程

一、创建RDD

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

1、读取外部数据：

val lines = sc.textFile("/path/to/README.md")

2、对集合进行并行化

val lines = sc.parallelize(List("pandas", "i like pandas"))

二、RDD操作

RDD支持两种操作：转化操作和行动操作。RDD的转化操作是返回一个新的RDD的操作，比如map()和filter()，而行动操作则是向驱动器程序返回结果或把结果写入外部系统的操作，会触发实际的计算，比如count()和first()。

如果对于一个特定的函数是属于转化操作还是行动操作感到困惑，你可以看看它的返回值类型：转化操作返回的是RDD，而行动操作返回的是其他的数据类型。

2.1、转化操作

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

2.2、行动操作

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

2.3、惰性求值

上面提过，RDD的转化操作都是惰性求值的。这意味着在被调用行动操作之前Spark不会开始计算。惰性求值意味着当我们对RDD调用转化操作（例如调用map()）时，操作不会立即执行。相反，Spark会在内部记录下所要求执行的操作的相关信息。我们不应该把RDD看作存放着特定数据的数据集，而最好把每个RDD当作我们通过转化操作构建出来的、记录如何计算数据的指令列表。把数据读取到RDD的操作同样也是惰性的。

三、常见的转化操作和行动操作

3.1、基本RDD

3.1.1、针对各个元素的转化操作

你很可能会用到的两个最常用的转化操作是map()和filter()。

1）map()

转化操作map()接收一个函数，把这个函数用于RDD中的每个元素，将函数的返回结果作为结果RDD中对应元素的值。

让我们看一个简单的例子，用map()对RDD中的所有数求平方。

val input = sc.parallelize(List(1,2,3,4))
val result = input.map(x => x * x)
println(result.collect().mkString(","))

2）flatMap()

有时候，我们希望对每个输入元素生成多个输出元素。实现该功能的操作叫作flatMap()。和map()类似，我们提供给flatMap()的函数被分别应用到了输入RDD的每个元素上。

3）filter()

转化操作filter()则接收一个函数，并将RDD中满足该函数的元素放入新的RDD中返回。

val lines = sc.parallelize(List("hello world", "hi"))
val words = lines.flatMap(line => line.split(" "))
words.first()

flatMap()和map()的区别：

你可以把flatMap()看作将返回的迭代器“拍扁”，这样就得到了一个由各列表中的元素组成的RDD，而不是一个由列表组成的RDD。

3.1.2、伪集合操作

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

4）distinct()

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

val RDD1 = sc.parallelize(Array("coffee","coffee","panda","monkey","tea"))
val OUTPUT = RDD1.distinct()
// 结果为"coffee","panda","monkey","tea"

 5）union()

最简单的集合操作是union(other)，它会返回一个包含两个RDD中所有元素的RDD。这在很多用例下都很有用，比如处理来自多个数据的日志文件。

val RDD1 = sc.parallelize(Array("coffee","coffee","panda","monkey","tea"))
val RDD2 = sc.parallelize(Array("coffee","monkey","kitty"))
val OUTPUT = RDD1.union(DDR2)
// 结果为"coffee","panda","monkey","tea","coffee","monkey","kitty"

 6）intersection()

Spark还提供了insertsection(other)方法，只返回两个RDD中都有的元素。insertsection()在运行时也会去掉所有重复的元素（单个RDD内的重复元素也会一起移除）。因为它需要通过网络混洗数据来发现共有的元素，所以它的性能比较差。

val RDD1 = sc.parallelize(Array("coffee","coffee","panda","monkey","tea"))
val RDD2 = sc.parallelize(Array("coffee","monkey","kitty"))
val OUTPUT = RDD1.intersection(DDR2)
// 结果为"coffee","monkey"

 7）subtract()

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

val RDD1 = sc.parallelize(Array("coffee","coffee","panda","monkey","tea"))
val RDD2 = sc.parallelize(Array("coffee","monkey","kitty"))
val OUTPUT = RDD1.subtract(DDR2)
// 结果为"panda","tea"

 8）cartesian()

我们也可以计算两个RDD的笛卡尔积。cartesian(other)转化操作会返回所有可能的(a,b)对，其中a是源RDD中的元素，而b则来自另一个RDD。

3.1.3、行动操作

1）reduce()

它接收一个函数作为参数，这个函数要操作两个RDD的元素类型的数据并返回一个同样类型的新元素。一个简单的例子就是函数+，可以用它来对我们的RDD进行累加。

val sum = rdd.reduce((x,y)=>x+y)

 再看一个例子，给定一个RDD[Int]求平均数：

// 创建一个RDD
val rdd1 = sc.parallelize(List(1,5,2,7,3,1,3,9,10))
// 使用map()将rdd1转化为二元组
val rdd2 = rdd1.map(x => (x, 1))
// 使用reduce()计算总和和总计数
val rdd3 = rdd2.reduce((x, y) => (x._1 + y._1, x._2 + y._2))
// 求平均值
val rdd4 = rdd3._1 / rdd3._2.toDouble