spark教程(四)-SparkContext 和 RDD 算子

SparkContext

SparkContext 是在 spark 库中定义的一个类，作为 spark 库的入口点；

它表示连接到 spark，在进行 spark 操作之前必须先创建一个 SparkContext 的实例，并且只能创建一个；

利用 SparkContext 实例创建的对象都是 RDD，这是相对于 SparkSession 说的，因为 它创建的对象都是 DataFrame；

创建 sc

class SparkContext(__builtin__.object):
    def __init__(self, master=None, appName=None, sparkHome=None, pyFiles=None, environment=None, batchSize=0, serializer=PickleSerializer(), conf=None, gateway=None, jsc=None, profiler_cls=<cl
    ass 'pyspark.profiler.BasicProfiler'>)
    '''Create a new SparkContext. At least the master and app name should be set,
     |      either through the named parameters here or through C{conf}.
     |
     |      :param master: Cluster URL to connect to
     |             (e.g. mesos://host:port, spark://host:port, local[4]).   local 表示本地运行，4 表示使用4个 cpu核
     |      :param appName: A name for your job, to display on the cluster web UI.
     |      :param sparkHome: Location where Spark is installed on cluster nodes.
     |      :param pyFiles: Collection of .zip or .py files to send to the cluster
     |             and add to PYTHONPATH.  These can be paths on the local file
     |             system or HDFS, HTTP, HTTPS, or FTP URLs.
     |      :param environment: A dictionary of environment variables to set on
     |             worker nodes.
     |      :param batchSize: The number of Python objects represented as a single
     |             Java object. Set 1 to disable batching, 0 to automatically choose
     |             the batch size based on object sizes, or -1 to use an unlimited
     |             batch size
     |      :param serializer: The serializer for RDDs.
     |      :param conf: A L{SparkConf} object setting Spark properties.
     |      :param gateway: Use an existing gateway and JVM, otherwise a new JVM
     |             will be instantiated.
     |      :param jsc: The JavaSparkContext instance (optional).
     |      :param profiler_cls: A class of custom Profiler used to do profiling
     |             (default is pyspark.profiler.BasicProfiler).
     |
     |
     |      >>> from pyspark.context import SparkContext
     |      >>> sc = SparkContext('local', 'test')
     |
     |      >>> sc2 = SparkContext('local', 'test2') # doctest: +IGNORE_EXCEPTION_DETAIL
     |      Traceback (most recent call last):
     |          ...
     |      ValueError:...'''

示例代码

from pyspark import SparkContext, SparkConf
### method 1
conf = SparkConf().setAppName('myapp1').setMaster('local[4]')   # 设定 appname 和 master
sc = SparkContext(conf=conf)

## method 2
sc = SparkContext("spark://hadoop10:7077")

创建 RDD

spark 最重要的一个概念叫 RDD，Resilient Distributed Dataset，弹性分布式数据集

spark 是以 RDD 概念为中心运行的，RDD 是一个容错的、可以被并行操作的元素集合。

创建 RDD 有三种方式：

1. 在驱动程序中并行化一个已经存在的集合　　　　【内存中的数据】

2. 从外部存储系统引入数据，生成 RDD　　　　　　【外部存储介质中的数据，注意 spark 本身没有存储功能】

　　// 这个存储系统可以是一个共享文件系统，如 hdfs、hbase

3. 从一种 RDD 转换成 另一种 RDD

详见我的博客 RDD 认知

操作 RDD

RDD 的操作有两种方式：转换 和 行动，而且 转换 是 惰性的

可以根据 是否有返回 判断是哪个操作，行动 有返回值，转换无返回值

详见官网 RDD

RDD 的操作也叫 RDD 算子 

转换 算子

惰性，无返回值

map(func[, preservesPartitioning=False])：把一个序列中的元素逐个送入 map，经 func 处理后，返回一个新的 序列

rdd = sc.parallelize([2, 3, 4])
rdd.map(lambda x: x + 1).collect()          # [3, 4, 5]

filter(func)：类似 map，func 是个过滤函数

rdd = sc.parallelize([2, 3, 4])
rdd.map(lambda x: x > 3).collect()          # [False, False, True]

flatMap(func[, preservesPartitioning=False])：也类似 map，只是 它会把 每次经过 func 处理的结果进行 合并，输入和输出的 list 长度可能不同

rdd = sc.parallelize([2, 3, 4])
rdd.flatMap(lambda x: range(1, x)).collect()            # [1, 1, 2, 1, 2, 3]
# range(1, 2): 1
# range(1, 3): 1, 2
# range(1, 4): 1, 2, 3

### vs map
rdd.map(lambda x: range(1, x)).collect()                # [[1], [1, 2], [1, 2, 3]]

mapPartitions(func [, preservesPartitioning=False]) ：map的一个变种，map 是把序列的单个元素送入 func ，而 mapPartitions 是把 序列分区后 每个 分区 整体送入 func

rdd = sc.parallelize([1,2,3,4,5], 3)    # 分 3 个区
def f(iterator): yield sum(iterator)    # 必须是生成器，即 yield，不能 return
rdd.mapPartitions(f).collect()          # [1, 5, 9]

mapPartitionsWithIndex(func [, preservesPartitioning=False]) ：func 有两个参数，分片的序号 和 迭代器，返回 分片序号，也必须是 迭代器

rdd = sc.parallelize(range(15), 13)    # 分 13 个区
def f(splitIndex, iterator): yield splitIndex
rdd.mapPartitionsWithIndex(f).collect() # [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12]

cartesian(otherDataset)：利用两个 序列 生成 笛卡尔內积 的数据集

x = sc.parallelize([1,2,3])
y = sc.parallelize([4,5])
x.cartesian(y).collect()        # [(1, 4), (1, 5), (2, 4), (2, 5), (3, 4), (3, 5)]

groupBy(self, f, numPartitions=None, partitionFunc=portable_hash)：输入一个 func，每个元素都用经过 func 处理，并对处理后的结果自动分组

## 条件是分组依据，条件不影响最后的输出格式，输出格式仍和原数据相同
## 如 原来是 [1， 2]，经过分组后分到了 第 1 组，输出是 [1, [1， 2]], [1, 2] 完全保留

# 这个例子相当于求 奇偶数
rdd = sc.parallelize([1, 1, 2, 3, 5, 8])
result = rdd.groupBy(lambda x: x % 2).collect()     # [(0, <pyspark.resultiterable.ResultIterable object at 0x7f2f76096890>),
                                                    # (1, <pyspark.resultiterable.ResultIterable object at 0x7f2f760965d0>)]
# 解析迭代器并排序
sorted([(x, sorted(y)) for (x, y) in result])       # [(0, [2, 8]), (1, [1, 1, 3, 5])]

keyBy(self, f)：利于 func 为 rdd 中每个元素设置一个 key

x = sc.parallelize(range(0,3)).keyBy(lambda x: x*x).collect()
# [(0, 0), (1, 1), (4, 2)]

sortBy(self, keyfunc, ascending=True, numPartitions=None)：根据指定函数进行排序

tmp = [('a', 1), ('b', 2), ('1', 3), ('d', 4), ('2', 5)]
sc.parallelize(tmp).sortBy(lambda x: x[0]).collect()
# [('1', 3), ('2', 5), ('a', 1), ('b', 2), ('d', 4)]
sc.parallelize(tmp).sortBy(lambda x: x[1]).collect()
# [('a', 1), ('b', 2), ('1', 3), ('d', 4), ('2', 5)]

cogroup(self, other, numPartitions=None)：把 kv 对按 k 进行收集

x = sc.parallelize([("a", 1), ("b", 4)])
y = sc.parallelize([("a", 2)])
[(x, tuple(map(list, y))) for x, y in # sorted(list(x.cogroup(y).collect()))]
[('a', ([1], [2])), ('b', ([4], []))]

以下方法只适用 key-value 数据

keys(self)，values(self)

m = sc.parallelize([(1, 2), (3, 4)]).keys()
m.collect()    # [1, 3]

mapValues(func)：根据 func 处理 value

rdd = sc.parallelize([(1, [1,2,3]), (3, ['a', 'b'])])
rdd.mapValues(len).collect()            # [(1, 3), (3, 2)]  计算 value 的长度

reduceByKey(func [, numPartitions=None, partitionFunc=<function portable_hash at 0x7fa664f3cb90>])：针对 k-v 对的处理方法，把 key 相同的 value 进行 reduce，然后重新组成 key-reduce 对

rdd = sc.parallelize([("a", 1), ("b", 1), ("a", 1)])
def f(x, y): return x + y
rdd.reduceByKey(f).collect()        # [('a', 2), ('b', 1)]

groupByKey(self, numPartitions=None, partitionFunc=portable_hash)：根据 key 进行分组

 rdd.groupByKey().collect()
# [('a', <pyspark.resultiterable.ResultIterable object at 0x7f16e3c0bd90>), ('b', <pyspark.resultiterable.ResultIterable object at 0x7f16e3c0bcd0>)]
sorted(rdd.groupByKey().mapValues(len).collect())
# [('a', 2), ('b', 1)]
sorted(rdd.groupByKey().mapValues(list).collect())
# [('a', [1, 1]), ('b', [1])]

sortByKey([ascending=True, numPartitions=None, keyfunc=<function <lambda> at 0x7fa665048c80>])：根据 key 进行排序，默认升序，numPartitions 代表分区数，keyfunc 是处理 key 的，在 排序过程中对 key 进行处理

tmp = [('a', 4), ('b', 3), ('c', 2), ('D', 1)]
sc.parallelize(tmp).sortByKey(True, 1).collect()    # 升序[('D', 1), ('a', 4), ('b', 3), ('c', 2)] 1代表分区数
sc.parallelize(tmp).sortByKey(True, 2, keyfunc=lambda k:k.lower()).collect()   # 升序[('a', 4), ('b', 3), ('c', 2), ('D', 1)] D跑到后面了

sc.parallelize(tmp).sortByKey(False, 2, keyfunc=lambda k:k.lower()).collect()# 降序[('D', 1), ('c', 2), ('b', 3), ('a', 4)]

keyfunc 只在 排序过程中起作用，在输出时 keyfunc 不起作用

join(otherDataset [, numPartitions=None])：将 两个 k-v RDD 中 共有的 key 的 value 交叉组合

x = sc.parallelize([("a", 1), ("b", 4)])
y = sc.parallelize([("a", 2), ("a", 3)])
x.join(y).collect()     # [('a', (1, 2)), ('a', (1, 3))]

join 是内连接，也就是共有的 key 进行组合；

还有 leftOuterJoin 左连接，rightOuterJoin 右连接，fullOuterJoin 全连接

groupWith(other, *others)：把多个 RDD 的 key 进行分组；输出 (key，迭代器)

分组后的数据是有顺序的，每个 key 对应的 value 是按 原本 RDD 的顺序的，如果原本 RDD 没有这个 key，留空

w = sc.parallelize([("a", 5), ("b", 6)])
x = sc.parallelize([("a", 1), ("b", 4)])
y = sc.parallelize([("a", 2)])
z = sc.parallelize([("b", 42)])
w.groupWith(x, y, z).collect()

[(x, tuple(map(list, y))) for x, y in list(w.groupWith(x, y, z).collect())]     # [('a', ([5], [1], [2], [])), ('b', ([6], [4], [], [42]))]

行动 算子

有返回值

count：返回 RDD 中元素个数

first：返回 RDD 中第一个元素

max. min.sum：不解释

take(n)：返回 RDD 中 前 n 个元素

collect：返回 RDD 中的数据

要注意的是，collect 相当于把 worker 的数据转移给驱动程序来显示，如果数据量过大，可能导致驱动程序崩溃

takeOrdered(n [, key=None])：对 RDD 先进行排序，然后取排序后的 前 n 个数据，key 表示先经过 keyfunc 处理后再进行排序，最终返回的还是原数据

sc.parallelize([9,7,3,2,6,4]).takeOrdered(3)    # [2, 3, 4]
sc.parallelize([9,7,3,2,6,4]).takeOrdered(3, key=lambda x: -x)   # [9, 7, 6]
## 过程如下
#  9,  7,  3,  2,  6,  4  ## 原数据
# -9, -7, -3, -2, -6, -4  ## 经过 keyfunc 处理后的数据
# -9, -7, -6, -4, -3, -2  ## 对处理后的数据升序排序
# -9, -7, -6              ## 取前3个
#  9,  7,  6              ## 对应到原数据

也就是说，keyfunc 只在排序时起作用，在输出时不起作用

foreach(func)：运行 func 函数 并行处理 RDD 的所有元素

它与 map 的区别在于一个是 转换算子，一个是行动算子　　　　【曾经面试被问过这个问题，从功能上讲两个函数好像一样的，当时没答上，心碎】

sc.parallelize([1, 2, 3, 4, 5]).foreach(print)  # 并行打印，不按顺序输出
# 1
# 2
# 4
# 5
# 3

reduce(func)：把 RDD 中前两个元素送入 func，得到一个 value，把这个 value 和 下一个元素 送入 func，直至最后一个元素

sc.parallelize([1,2,3,4,5]).reduce(lambda x, y: x + y)  # 15 求和

fold：与 reduce 类似，fold 是有一个 基数，然后 把每个元素 和 基数 送入 func，然后替换该基数，循环，直到最后一个元素

x = sc.parallelize([1,2,3])
neutral_zero_value = 0  # 0 for sum, 1 for multiplication
y = x.fold(neutral_zero_value, lambda obj, accumulated: accumulated + obj) # computes cumulative sum
print(x.collect())  # [1,2,3]
print(y)            # 6

aggregate：对每个分区进行聚合，然后聚合每个分区的聚合结果，详见我的博客 aggregate

countByValue：统计相同元素的个数

sc.parallelize([1,2,3,1,2,5,3,2,3,2]).countByValue().items()    # [(1, 2), (2, 4), (3, 3), (5, 1)]

# 输入 k-v 不按 value 统计，按 k-v 统计
sc.parallelize([('a', 1), ('b', 1)]).countByValue().items()     # [(('a', 1), 1), (('b', 1), 1)]

saveAsTextFile(path [, compressionCodecClass=None])：把 RDD 存储到文件系统中

counts.saveAsTextFile('/usr/lib/spark/out')

输入必须是 路径，且该路径不能事先存在

以下方法只适用 key-value 数据

countByKey：统计相同 key 的个数，返回 key-count 

sc.parallelize([("a",1), ("b",1), ("a", 3)]).countByKey()   # defaultdict(<type 'int'>, {'a': 2, 'b': 1})

dictdata= sc.parallelize([("a",1), ("b",1), ("a", 3)]).countByKey()
dictdata.items()        # [('a', 2), ('b', 1)]

lookup(self, key)：输入 key，返回对应的所有 value

rdd = sc.parallelize([('a', 1), ('b',3), ('a', 5)])
rdd.lookup('a')    # [1, 5]

RDD 缓存

把 RDD 缓存到 内存中， 惰性执行

distFile = sc.textFile('README.md')
m = distFile.map(lambda x: len(x))      # map 是 转换 操作，并不立即执行
m.cache()       # 把 map 的输出缓存到内存中，其实 cache 就是 执行 操作
# 或者 m.persist()

缓存除了存储到内存中，还有一个非常重要的作用，复用以提高效率；

缓存的 RDD 不仅可以直接拿来转换成新的 RDD，还可以多次利用；

我们知道 RDD 是有血缘关系的，即一个 RDD 由另一个 RDD 转换得来，而这种关系可能是多层的；

如果一个 RDD 缓存了，spark 会执行到目前为止所有转换操作，并为生成的 RDD 创建一个检查点，

但是由于 缓存 是惰性操作，缓存只会在第一次 行动 操作后创建，且第一次 行动 操作不受益，第二次三次直接调用缓存才受益；

缓存适合多次使用的数据，只用一次的无需缓存；

缓存适合大数据，小数据无需缓存

persist(self, storageLevel=StorageLevel.MEMORY_ONLY)：这个方法加入了 存储等级，可以把 RDD 缓存在内存或者磁盘上，参数可选，如果没有，等于 cache，存于内存中

参数取值： MEMORY_ONLY、DISK_ONLY、MEMORY_AND_DISK 等

RDD 缓存可容错，即有备份的；

RDD 保存

保存到文件等存储介质中，行动操作

saveAsNewAPIHadoopDataset(self, conf, keyConverter=None, valueConverter=None)

saveAsNewAPIHadoopFile(self, path, outputFormatClass, keyClass=None, valueClass=None, keyConverter=None, valueConverter=None, conf=None)

saveAsHadoopDataset(self, conf, keyConverter=None, valueConverter=None)

saveAsHadoopFile(self, path, outputFormatClass, keyClass=None, valueClass=None, keyConverter=None, valueConverter=None, conf=None, compressionCodecClass=None)

saveAsSequenceFile(self, path, compressionCodecClass=None)

saveAsPickleFile(self, path, batchSize=10)

saveAsTextFile(self, path, compressionCodecClass=None)

参考资料：

https://www.cnblogs.com/yangzhang-home/p/6056133.html　　快速入门

https://blog.csdn.net/kl28978113/article/details/80361452　　较全教程

http://spark.apache.org/docs/latest/　　　　spark 2.4.4 官网

http://spark.apache.org/docs/latest/api/python/index.html  　　spark 2.4.4 python API

https://www.cnblogs.com/Vito2008/p/5216324.html

https://blog.csdn.net/proplume/article/details/79798289

https://www.iteblog.com/archives/1396.html#aggregate　　RDD 操作 API

https://www.cnblogs.com/yxpblog/p/5269314.html　　　　RDD 操作 API