RDD（Resilient Distributed Dataset），全称弹性分布式数据集，是Spark对数据进行的核心抽象概念。我们可以将RDD理解为一个不可变的分布式对象集合，他可以包含Python、Java、Scala 中任意类型的对象，甚至是用户自定义的对象。Spark中的所有操作都是在RDD进行的，包括创建RDD，转化RDD跟调用RDD。

RDD创建

Spark有两种方法创建RDD：读取一个外部数据集，或在从程序中读取一个对象集合（比如list 和set）。

# 读取一个外部数据集
lines = sc.textFile("E:\Documents\Desktop\s.txt")
# 读取一个对象集合
lines = sc.parallelize([1, 2, 3, 4])

RDD操作

RDD 支持两种类型的操作： 转化操作（transformation） 和行动操作（action）。转化操作会由一个RDD 生成一个新的RDD。行动操作是对的RDD 内容进行操作，它们会把最终求得的结果返回到驱动器程序，或者写入外部存储系统中。由于行动操作需要生成实际的输出，它们会强制执行那些求值必须用到的RDD 的转化操作。

RDD的转化操作与行动操作不同，是惰性求值的，也就是在被调用行动操作之前Spark 不会开始计算。同样创建操作也是一样，数据并没有被立刻读取到内存中，只是记录了读取操作需要的相关信息。我理解为这与tensorflowde的网络构建类似，我们之前编写的代码只是记录了整个操作过程的计算流程图，只有当计算操作被激活时，数据才会沿着之前定义的计算图进行计算。

转化

许多转化操作都是针对元素的，也就是说这些转化操作每次只会操作RDD 中的一个元素。RDD的转化操作函数接受一个函数，可以是匿名函数也可以是自定义函数，然后返回对每个元素进行操作后的RDD。

filter()

# 返回一个由通过传给filter()的函数的元素组成的RDD
>>> rdd = sc.parallelize([1, 2, 3, 4])
>>> rdd2 = rdd.filter(lambda x: x > 1)
>>> rdd2.collect()
[2, 3, 4]

map()

# 将函数应用于RDD 中的每个元素，将返回值构成新的RDD
>>> rdd = sc.parallelize([1, 2, 3, 4])
>>> rdd2 = rdd.map(lambda x: x + 1)
>>> rdd2.collect()
[2, 3, 4, 5]

flatMap()

# 将函数应用于RDD 中的每个元素，将返回的迭代器的所有内容构成新的RDD。通常用来切分单词。
>>> rdd = sc.parallelize(['ad cd', 'ef gh', 'hi jk'])
>>> rdd2 = rdd.flatMap(lambda x: x.split(' '))
>>> rdd2.collect()
['ad', 'cd', 'ef', 'gh', 'hi', 'jk']

distinct()

# 去重
>>> rdd = sc.parallelize([2, 2, 3, 4, 5, 5])
>>> rdd2 = rdd2.distinct()
>>> rdd2.collect()
[4, 5, 2, 3]

union()

# 生成一个包含两个RDD 中所有元素的RDD
>>> rdd = sc.parallelize([1, 2, 3, 4])
>>> rdd2 = sc.parallelize([2, 2, 3, 4, 5, 5])
>>> rdd3 = rdd.union(rdd2)
>>> rdd3.collect()
[1, 2, 3, 4, 2, 2, 3, 4, 5, 5]

intersection()

# 求两个RDD 共同的元素的RDD
>>> rdd = sc.parallelize([1, 2, 3, 4])
>>> rdd2 = sc.parallelize([2, 2, 3, 4, 5, 5])
>>> rdd3 = rdd.intersection(rdd2)
>>> rdd3.collect()
[2, 3, 4]

subtract()

# 移除一个RDD 中的内容
>>> rdd = sc.parallelize([1, 2, 3, 4])
>>> rdd2 = sc.parallelize([2, 2, 3, 4, 5, 5])
>>> rdd3 = rdd.subtract(rdd2)
>>> rdd3.collect()
[1]

cartesian()

# 与另一个RDD 的笛卡尔积
>>> rdd = sc.parallelize([1, 2, 3, 4])
>>> rdd2 = sc.parallelize([2, 3, 4, 5])
>>> rdd3 = rdd.cartesian(rdd2)
>>> rdd3.collect()
[(1, 2), (1, 3), (1, 4), (1, 5), (2, 2), (2, 3), (2, 4), (2, 5), (3, 2), (3, 3), (3, 4), (3, 5), (4, 2), (4, 3), (4, 4), (
4, 5)]

行动

行动操作对RDD进行计算，并把结果返回到驱动器程序中。

collect()

# 返回RDD 中的所有元素
>>> rdd = sc.parallelize([1, 2, 3, 4])
>>> rdd.collect()
[1, 2, 3, 4]

count()

# RDD 中的元素个数
>>> rdd = sc.parallelize([1, 2, 3, 4])
>>> rdd.count()
4

countByValue()

# 各元素在RDD 中出现的次
>>> rdd = sc.parallelize([2, 2, 3, 4, 5, 5])
>>> rdd.countByValue()
defaultdict(<class 'int'>, {2: 2, 3: 1, 4: 1, 5: 2})
>>>

take()

# 从RDD 中返回num 个元素
>>> rdd = sc.parallelize([2, 2, 3, 4, 5, 5])
>>> rdd.take(2)
[2, 2]

top()

从RDD 中返回最前面的num个元素
>>> rdd = sc.parallelize([2, 2, 3, 4, 5, 5])
>>> rdd.top(3)
[5, 5, 4]

takeOrdered()

# 从RDD 中按照默认（升序）或指定排序规则，返回前num个元素。
>>> rdd = sc.parallelize([2, 2, 3, 4, 5, 5])
>>> rdd.takeOrdered(3)
[2, 2, 3]

reduce()

# 并行整合RDD 中所有数据
>>> rdd = sc.parallelize([2, 2, 3, 4, 5, 5])
>>> rdd.reduce(lambda x, y: x + y)
21

fold()

# 和reduce() 一样， 但是需要提供初始值
>>> rdd = sc.parallelize([2, 2, 3, 4, 5, 5])
>>> rdd.fold(0, lambda x, y: x + y)
21

aggregate()

# 和reduce() 相似， 但是通常返回不同类型的函数
>>> seqOp = (lambda x, y: (x[0] + y, x[1] + 1))
>>> combOp = (lambda x, y: (x[0] + y[0], x[1] + y[1]))
>>> sc.parallelize([1, 2, 3, 4]).aggregate((0, 0), seqOp, combOp)
(10, 4)

foreach()

# 对RDD 中的每个元素使用给定的函数，但是不把任何结果返回到驱动器程序中。

持久化

Spark RDD 是惰性求值的，而有时我们希望能多次使用同一个RDD。如果简单地对RDD 调用行动操作，Spark 每次都会重算RDD 以及它的所有依赖。这在迭代算法中消耗格外大，因为迭代算法常常会多次使用同一组数据。

# 提供了不同类型的持久化参数，可以传递到persist()中
pyspark.StorageLevel

# 持久化
rdd.persist

Tips

1.使用RDD.distinct() 转化操作来生成一个只包含不同元素的新RDD时需要注意，distinct() 操作的开销很大，因为它需要将所有数据通过网络进行混洗（shuffle），以确保每个元素都只有一份。此外subtract()等其他集合运rdd算中，所有使用了去重的函数都会进行混洗。

2.Python 会在把函数所在的对象也序列化传出去。当传递的对象是某个对象的成员，或者包含了对某个对象中一个字段的引用时（例如self.field），Spark 就会把整个对象发到工作节点上。