3.5 常见的转化操作和行动操作

　　3.5.1 基本RDD

　　　　1. 针对各个元素的转化操作

　　　　两个最常用的转化操作是map()和filter()。转化操作map()接受一个函数，把这个函数用于RDD中的每个元素，将函数的返回结果作为结果RDD中对应元素的值。而转化操作filter()则接收一个函数，并将RDD中满足该函数的元素放入新的RDD中返回。

　　　　inputRDD{1，2，3，4} >>>map x=>x*x >>> Mapped RDD{1，4，9，16}

　　　　inputRDD{1，2，3，4} >>>filter x=>x!=1 >>> Filtered RDD{2，3，4}

　　　　例 3-26：Python版计算RDD中各值的平方

nums = sc.parallelize([1,2,3,4])

squared = nums.map(lambda x : x*x)

for num in squared:

    print "%i "%(num)

　　　　例 3-27：Scala版计算RDD中各值的平方

val input = sc.parallelize(List(1,2,3,4))

val result = input.map(x => x*x)

println(result.collect().mkString(","))

　　　　例 3-29： Python中的flatMap（）将行数据切分为单词

>>> lines = sc.parallelize(["hello world","hi"])

>>> words1 = lines.flatMap(lambda line: line.split(" "))

>>> words2 = lines.map(lambda line : line.split(" "))

>>> words1.first()

'hello'

>>> words2.first()

['hello', 'world']

　　　　例 3-30： Scala中的flatMap（）将行数据切分为单词

val lines = sc.parallelize(["hello world", "hi"])

words = lines.flatMap(line=> line.split(" "))

words.first()

　　　　2. 伪集合操作

　　　　RDD支持许多数学上的集合操作，比如合并和相交。注意，这些操作都要求操作的RDD、是相同数据类型的。

　　　　RDD.distinct()生成一个只包含不同元素的新RDD。

　　　　union(other)操作返回一个包含两个RDD中所有元素的RDD。

　　　　intersection（other）方法只返回两个RDD中都有的元素。

　　　　subtract（other）函数接受另一个RDD作为参数，返回一个由只存在于第一个RDD中而不存在第二个RDD中的所有元素组成的RDD。

　　　　cartesian（other）转化操作会返回所有可能的（a，b）对，其中a是源RDD中的元素，b来自另一个RDD。

　　　　3. 行动操作

　　　　例 3-32：Python中的reduce（）

>>> nums = sc.parallelize([1,2,3,4])

>>> sums = nums.reduce(lambda x, y:x + y)

>>> sums

10

　　　　例3-33：Scala中的reduce（）

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

　　　　RDD的一些行动操作会以普通集合或者值的形式将RDD的部分或全部数据返回驱动器程序中。

　　　　collect（）操作会将整个RDD的内容返回。

　　　　take（n）返回RDD中的n个元素。

　　　　top（n）从RDD中获取前n个元素。

　　　　foreach（func）对RDD中的每个元素使用给定的函数。

　　　　count（）用来返回元素个数。

　　　　3.5.2 在不同RDD类型间转换

　　　　略

　　3.6 持久化（缓存）

　　　　Spark RDD是惰性求值的，而有时我们希望能多次使用同一个RDD。如果简单的对RDD调用行动操作，Spark每次都会重算RDD以及它的所有依赖。

　　　　例3-39： Scala中的两次执行

val result = input.map(x => x*x)

println(result.count())

println(result.collect().mkString(","))

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

　　　　例3-40： 在Scala中使用persist（）

val result = input.map(x => x*x)

result.persist(StorageLevel.DISK_ONLY)

println(result.count())

println(result.collect().mkString(","))

　　　　注意，persist()调用本身不会引发强制求值。

　　　　RDD还有一个方法叫做unpersist（），调用该方法可以手动把持久化的RDD从缓存中移除。