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
原文地址：https://www.cnblogs.com/liuys635/p/12208457.html