5. 分区相关的操作

分区的作用主要是增加并行度和减少通信开销。增加并行度就不需要多说了，我们主要来聊一聊减少通信开销。我看书上的意思大概是说分区之后避免了后续处理当中的Shuffle操作，意思是已经把数据给排列好了的感觉，所以说不用后面一次又一次地重新Shuffle。

分区默认的方法有HashPartitioner和RangePartitioner，前者是根据Key进行分区，后者是为了更好地排序。

5.1 改变分区的函数

5.1.1 coalesce

>>> rdd = sc.parallelize([0, 1, 2, 3, 4, 5, 6, 7, 8, 9],3) >>> rdd.glom().collect() [[0, 1, 2], [3, 4, 5], [6, 7, 8, 9]] >>> rdd = rdd.coalesce(2) >>> rdd.glom().collect() [[0, 1, 2], [3, 4, 5, 6, 7, 8, 9]] >>> rdd = rdd.coalesce(3, True) >>> rdd.glom().collect() [[], [0, 1, 2], [3, 4, 5, 6, 7, 8, 9]] 

coalesce默认的shuffle是False，只能减少分区的数量，如果想要增加分区可以调整为True。但我看其他博客好像太建议这个函数，都是说慎重操作。

5.1.2 repartition

>>> rdd = sc.parallelize([0, 1, 2, 3, 4, 5, 6, 7, 8, 9],3) >>> rdd.glom().collect() [[0, 1, 2], [3, 4, 5], [6, 7, 8, 9]] >>> rdd = rdd.repartition(2) >>> rdd.glom().collect() [[0, 1, 2, 6, 7, 8, 9], [3, 4, 5]] >>> rdd = rdd.repartition(3) >>> rdd.glom().collect() [[], [0, 1, 2], [6, 7, 8, 9, 3, 4, 5]] >>> rdd = sc.parallelize([ ... ('Mickey', 0), ... ('Mickey', 1), ... ('Mickey', 2), ... ('Minnie', 3), ... ('Minnie', 4), ... ('Pluto', 5), ... ('Goofy', 6), ... ('Goofy', 7), ... ('Donald', 8) ... ], 3) >>> pprint(rdd.glom().collect()) [ [('Mickey', 0), ('Mickey', 1), ('Mickey', 2)], [('Minnie', 3), ('Minnie', 4), ('Pluto', 5)], [('Goofy', 6), ('Goofy', 7), ('Donald', 8)] ] >>> rdd = rdd.repartition(2) >>> pprint(rdd.glom().collect()) [ [ ('Mickey', 0), ('Mickey', 1), ('Mickey', 2), ('Goofy', 6), ('Goofy', 7), ('Donald', 8) ], [('Minnie', 3), ('Minnie', 4), ('Pluto', 5)] ] >>> rdd = rdd.repartition(3) >>> pprint(rdd.glom().collect()) [ [], [('Mickey', 0), ('Mickey', 1), ('Mickey', 2)], [ ('Goofy', 6), ('Goofy', 7), ('Donald', 8), ('Minnie', 3), ('Minnie', 4), ('Pluto', 5) ] ] >>> 

该函数好像就是调用coalesce实现的操作，由于shuffle可以设定为True所以说能够增加分区的数量。具体分区的操作是按随机数进行，相同的Key不一定在同一个分区。

5.1.3 partitionBy

>>> rdd = sc.parallelize([ ... ('Mickey', 0), ... ('Mickey', 1), ... ('Mickey', 2), ... ('Minnie', 3), ... ('Minnie', 4), ... ('Pluto', 5), ... ('Goofy', 6), ... ('Goofy', 7), ... ('Donald', 8) ... ], 3) >>> pprint(rdd.glom().collect()) [ [('Mickey', 0), ('Mickey', 1), ('Mickey', 2)], [('Minnie', 3), ('Minnie', 4), ('Pluto', 5)], [('Goofy', 6), ('Goofy', 7), ('Donald', 8)] ] >>> rdd = rdd.partitionBy(2) >>> pprint(rdd.glom().collect()) [ [ ('Mickey', 0), ('Mickey', 1), ('Mickey', 2), ('Minnie', 3), ('Minnie', 4), ('Goofy', 6), ('Goofy', 7) ], [('Pluto', 5), ('Donald', 8)] ] >>> rdd = rdd.partitionBy(4) >>> pprint(rdd.glom().collect()) [ [ ('Mickey', 0), ('Mickey', 1), ('Mickey', 2), ('Minnie', 3), ('Minnie', 4), ('Goofy', 6), ('Goofy', 7) ], [('Pluto', 5)], [], [('Donald', 8)] ] >>> 

该函数就是按Key来进行Shuffle，以此来保证相同的Key在同一个分区内。

该函数还可以传入一个函数作为第二个参数来接收数据的Key且输出对应的分区。

5.2 基于分区的函数

5.2.1 glom函数

>>> rdd = sc.parallelize([0, 1, 2, 3, 4, 5, 6, 7, 8, 9],2) >>> pprint(rdd.glom().collect()) [[0, 1, 2, 3, 4], [5, 6, 7, 8, 9]] >>> 

该函数的主要作用就是将各个分区内的数据转换为列表，然后再将不同分区的结果合并成列表。有返回结果，属于转换函数。

5.2.2 mapPartitions

>>> rdd = sc.parallelize([ ... ('Mickey', 0), ... ('Mickey', 1), ... ('Mickey', 2), ... ('Minnie', 3), ... ('Minnie', 4), ... ('Pluto', 5), ... ('Goofy', 6), ... ('Goofy', 7), ... ('Donald', 8) ... ], 3) >>> pprint(rdd.mapPartitions(lambda x:[(t[0], t[1]+1) for t in x]).glom().collect()) [ [('Mickey', 1), ('Mickey', 2), ('Mickey', 3)], [('Minnie', 4), ('Minnie', 5), ('Pluto', 6)], [('Goofy', 7), ('Goofy', 8), ('Donald', 9)] ] >>> pprint(rdd.mapPartitions(lambda x:list(x)).collect()) [ ('Mickey', 0), ('Mickey', 1), ('Mickey', 2), ('Minnie', 3), ('Minnie', 4), ('Pluto', 5), ('Goofy', 6), ('Goofy', 7), ('Donald', 8) ] >>> pprint(rdd.mapPartitions(lambda x:[list(x)]).collect()) [ [('Mickey', 0), ('Mickey', 1), ('Mickey', 2)], [('Minnie', 3), ('Minnie', 4), ('Pluto', 5)], [('Goofy', 6), ('Goofy', 7), ('Donald', 8)] ] >>> pprint(rdd.mapPartitions(lambda x:tuple(tuple(x))).collect()) [ ('Mickey', 0), ('Mickey', 1), ('Mickey', 2), ('Minnie', 3), ('Minnie', 4), ('Pluto', 5), ('Goofy', 6), ('Goofy', 7), ('Donald', 8) ] >>> pprint(rdd.mapPartitions(lambda x:list(tuple(x))).collect()) [ ('Mickey', 0), ('Mickey', 1), ('Mickey', 2), ('Minnie', 3), ('Minnie', 4), ('Pluto', 5), ('Goofy', 6), ('Goofy', 7), ('Donald', 8) ] 

这个函数的大概意思就是分别在各个分区上应用我们定义好的函数，比较神奇的是处理函数所接收的输入是迭代器，而且据博客所说其返回的数据也是迭代器。

在上面的第一个例子当中，我们对输入进行了解析，因为其是一个迭代器。至于输出，我们选择了列表形式，也是可迭代的对象。

后面的例子就有趣一些，我们对每个分区的输入进行列表化，也就是将迭代器转换为列表，然后发现结果聚集在了一起。

然后我们对已经是列表的结果，再一次进行列表化，这样就得到了按分区展示的结果。

所以我们可以大概推测，第一个列表化是将迭代器解压缩出数据，第二个列表化是将数据转为列表格式

但是如果我们选用组操作，就不能展示出分区的结果，挺神奇的。

该函数无返回结果，属于执行函数。

5.2.3 mapPartitionsWithIndex

>>> rdd = sc.parallelize([0, 1, 2, 3, 4, 5, 6, 7, 8, 9], 2) >>> pprint(rdd.glom().collect()) [[0, 1, 2, 3, 4], [5, 6, 7, 8, 9]] >>> def func(index, itor): ... if(index < 1): ... return list(map(lambda x:x+1, itor)) ... else: ... return list(map(lambda x:x+2, itor)) ... >>> rdd.mapPartitionsWithIndex(func).glom().collect() [[1, 2, 3, 4, 5], [7, 8, 9, 10, 11]] 

该函数和上面的函数有些类似，都是按照分区对数据进行处理。不同的是该函数传入了两个参数，一个是分区序号，另一个是分区内数据生成的迭代器。该函数无返回结果，属于执行函数。

我们可以借助TaskContext来查询素的分区序号，完成类似的操作。

>>> from pyspark.taskcontext import TaskContext >>> rdd = sc.parallelize([0, 1, 2, 3, 4, 5, 6, 7, 8, 9], 2) >>> def func(elem): ... index = TaskContext.get().partitionId() ... if(index < 1): ... return elem+1 ... else: ... return elem+2 ... >>> rdd.glom().collect() [[0, 1, 2, 3, 4], [5, 6, 7, 8, 9]] >>> rdd.map(func).glom().collect() [[1, 2, 3, 4, 5], [7, 8, 9, 10, 11]] >>> 

5.2.4 foreachPartitions

>>> rdd = sc.parallelize([0, 1, 2, 3, 4, 5, 6, 7, 8, 9], 2) >>> rdd.glom().collect() [[0, 1, 2, 3, 4], [5, 6, 7, 8, 9]] >>> rdd.foreachPartition(lambda x:print(max(x))) 4 9 

该函数以迭代器形式接收各个节点内的数据作为参数，无返回结果，属于执行函数。

5.2.5 aggregate

>>> def inner_func(t,x): ... print('t: {}, x: {}, t+x: {}'.format(t, x, t+x)) ... return t+x ... >>> def outer_func(p,q): ... print('p: {}, q: {}, p*q: {}'.format(p, q, p*q)) ... return p*q ... ... >>> rdd = sc.parallelize([0, 1, 2, 3, 4, 5, 6, 7, 8, 9], 2) >>> rdd.glom().collect() [[0, 1, 2, 3, 4], [5, 6, 7, 8, 9]] >>> rdd.aggregate(1, inner_func, outer_func) t: 1, x: 0, t+x: 1 t: 1, x: 1, t+x: 2 t: 2, x: 2, t+x: 4 t: 4, x: 3, t+x: 7 t: 7, x: 4, t+x: 11 t: 1, x: 5, t+x: 6 t: 6, x: 6, t+x: 12 t: 12, x: 7, t+x: 19 t: 19, x: 8, t+x: 27 t: 27, x: 9, t+x: 36 p: 1, q: 11, p*q: 11 p: 11, q: 36, p*q: 396 396 >>> 

这个函数接收三个参数：

第一个参数是初始值，无论是分区内的聚合还是分区间的聚合，都是把这个当作是初始值。

第二个参数是函数，决定分区内的聚合方式；第三个参数也是函数，决定分区间的聚合方式。

这个函数类似于foldByKey，因为它需要一个初始值，而且该函数是按照分区进行操作的。

5.2.6 aggregateByKey

>>> rdd = sc.parallelize([ ... ('Mickey', 0), ... ('Mickey', 1), ... ('Mickey', 2), ... ('Minnie', 3), ... ('Minnie', 4), ... ('Pluto', 5), ... ('Goofy', 6), ... ('Goofy', 7), ... ('Donald', 8) ... ], 4) >>> pprint(rdd.glom().collect()) [ [('Mickey', 0), ('Mickey', 1)], [('Mickey', 2), ('Minnie', 3)], [('Minnie', 4), ('Pluto', 5)], [('Goofy', 6), ('Goofy', 7), ('Donald', 8)] ] >>> from pyspark.taskcontext import TaskContext >>> def inner_func(t,x): ... currentId = TaskContext.get().partitionId() ... print('currentId: {}'.format(currentId)) ... print('t: {}, x: {}, t+x: {}'.format(t, x, t+x)) ... return t+x ... >>> def outer_func(p,q): ... print('p: {}, q: {}, p*q: {}'.format(p, q, p*q)) ... return p*q ... >>> pprint(rdd.aggregateByKey(0, inner_func, outer_func).collect()) currentId: 0 t: 0, x: 0, t+x: 0 currentId: 0 t: 0, x: 1, t+x: 1 currentId: 1 t: 0, x: 2, t+x: 2 currentId: 1 t: 0, x: 3, t+x: 3 currentId: 2 t: 0, x: 4, t+x: 4 currentId: 2 t: 0, x: 5, t+x: 5 currentId: 3 t: 0, x: 6, t+x: 6 currentId: 3 t: 6, x: 7, t+x: 13 currentId: 3 t: 0, x: 8, t+x: 8 p: 1, q: 2, p*q: 2 p: 3, q: 4, p*q: 12 [ ('Mickey', 2), ('Minnie', 12), ('Goofy', 13), ('Pluto', 5), ('Donald', 8) ] >>> 

这个函数和上面的aggregate有点类似，但是该函数是处理键值对类型数据的。而且该函数的初始值仅用于同一个分区内相同Key数据的归并，并不用于不同分区之间的数据合并。还有一个问题，那就是该函数属于转换函数，返回的结果是一个RDD，而且并不立即执行，这和作为执行函数的aggregate还是很不一样的。

6. 其他的一些操作

6.1 缓存操作

我们之前已经讲述过了，RDD具有惰性计算的特性，也就是直到遇到执行函数的时候才会进行计算。如果某一个RDD出现在了两条血缘关系当中，那就意味着它可能要被计算两次。这无疑是浪费时间和资源的，所以我们可以通过设置缓存来储存其第一次的计算结果，以方便第二次直接使用。需要注意的是，当我们在对RDD进行缓存设置时，它并不会被立即计算并缓存，而是等到第一次被计算的时候才缓存。同时该缓存设置并不会切断RDD本来的血缘依赖关系，因为当数据缓存发生故障的时候我们需要依据血缘关系来重新计算。

>>> rdd = sc.parallelize([0, 1, 2, 3, 4, 5, 6, 7, 8, 9]) >>> rdd.cache() ParallelCollectionRDD[75] at readRDDFromFile at PythonRDD.scala:274 >>> rdd.collect() [0, 1, 2, 3, 4, 5, 6, 7, 8, 9] >>> rdd.count() 10 >>> rdd.reduce(lambda x,y:x+y) 45 >>> from pyspark.storagelevel import StorageLevel >>> rdd = sc.parallelize(range(10000),5) >>> rdd.persist(StorageLevel.MEMORY_AND_DISK) PythonRDD[80] at RDD at PythonRDD.scala:53 >>> sum_rdd = rdd.reduce(lambda x,y:x+y) >>> cnt_rdd = rdd.count() >>> sum_rdd/cnt_rdd 4999.5 >>> rdd.unpersist() PythonRDD[80] at RDD at PythonRDD.scala:53 >>> 

较为简单的操作是cache()，它调用了persist()方法，并选择了MEMORY_AND_DISK持久化策略¹。在释放缓存的时候，我们可以使用unpersist()方法。

6.2 共享变量

我们知道Spark是一个分布式计算框架，默认情况下是会把分布式运行的函数所涉及到的对象在每个节点生成一个副本。但是不同的节点和任务之间存在变量共享的需要，一般有广播变量和累加器两种方式。

6.2.1 广播变量

广播变量是不可变的类型，主要用于在不同节点、不同任务之间共享数据。其主要是在每台机器上缓存一个只读变量，而不是为每个任务生成独立的副本。

>>> rdd = sc.parallelize([0, 1, 2, 3, 4, 5, 6, 7, 8, 9]) >>> broads = sc.broadcast(1) >>> rdd.map(lambda x:x+broads.value).collect() [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] >>> 

6.2.2 累加器

累加器是只在Driver节点上可读的，在其他节点上只能累加，是一个有力的工具²。

>>> rdd = sc.parallelize([0, 1, 2, 3, 4, 5, 6, 7, 8, 9]) >>> total = sc.accumulator(0) >>> count = sc.accumulator(0) >>> def func(x): ... total.add(x) ... count.add(1) ... >>> rdd.foreach(func) >>> total.value/count.value 4.5 >>> 

但是我们需要注意累加器的一些陷阱³：

首先是累加器也遵循懒惰计算的原则，如果我们直接使用累加器变量例如acc += 1，那么这段代码只有在遇到Action算子的时候才会进行真正的运算。

其次还是与懒惰计算有关，假如某一个RDD的血缘关系当中存在一个累加器变量，而这个RDD最终转换的结果并没有被缓存。那么每当我们遇到涉及这个RDD的执行算子，累加器变量都要执行一遍，也就是说可能会有重复计算的情况。同理，如果被缓存的RDD提前释放的话，也会造成重复计算的情况。

7. RDD编程练习

建议先去原文处⁴做题，这里贴的是我的练习，先贴一下建环境的代码：

import warnings warnings.filterwarnings('ignore') # 过滤掉提示 from prettyprinter import pprint # 美化打印输出 from prettytable import PrettyTable # 美化打印输出 from pyspark import SparkContext, SparkConf conf = SparkConf().setAppName("rdd_tutorial").setMaster("local") sc = SparkContext(conf=conf) # 创建Spark的上下文环境sc sc.setLogLevel("Error") # 设置日志的输出级别 

7.1 求平均数

>>> rdd = sc.parallelize([1,5,7,10,23,20,6,5,10,7,10]) >>> sum_rdd = rdd.reduce(lambda x,y:x+y) >>> cnt_rdd = rdd.count() >>> sum_rdd/cnt_rdd 9.5455 

答案的reduce函数当中多加了一个0.0，应该是为了转化成为浮点数吧。

7.2 求众数

>>> rdd = sc.parallelize([1,5,7,10,23,20,6,5,10,7,10]) >>> rdd = rdd.zipWithIndex() >>> rdd.countByKey() defaultdict(<class 'int'>, { 
   1: 1, 5: 2, 7: 2, 10: 3, 23: 1, 20: 1, 6: 1}) 

这里没看清题目要求，如果有多个众数就求平均值，此外代码应该直接输出众数才是，贴一下答案：

#任务：求data中出现次数最多的数，若有多个，求这些数的平均值
data = [1,5,7,10,23,20,7,5,10,7,10]
rdd_data = sc.parallelize(data)
rdd_count = rdd_data.map(lambda x:(x,1)).reduceByKey(lambda x,y:x+y)
max_count = rdd_count.map(lambda x:x[1]).reduce(lambda x,y: x if x>=y else y)
rdd_mode = rdd_count.filter(lambda x:x[1]==max_count).map(lambda x:x[0])
mode = rdd_mode.reduce(lambda x,y:x+y+0.0)/rdd_mode.count()
print(“mode:”,mode)

7.3 求TopN

有一批学生信息表格，包括name,age,score, 找出score排名前3的学生, score相同可以任取

>>> rdd = sc.parallelize([("LiLei",18,87),("HanMeiMei",16,77),("DaChui",16,66),("Jim",18,77),("RuHua",18,50)]) >>> pprint(rdd.sortBy(lambda x:x[2],False).collect()) [ ('LiLei', 18, 87), ('HanMeiMei', 16, 77), ('Jim', 18, 77), ('DaChui', 16, 66), ('RuHua', 18, 50) ] >>> 

代码应该直接输出TopN，再来一步take函数，答案如下：

#任务：有一批学生信息表格，包括name,age,score, 找出score排名前3的学生, score相同可以任取
students = [(“LiLei”,18,87),(“HanMeiMei”,16,77),(“DaChui”,16,66),(“Jim”,18,77),(“RuHua”,18,50)]
n = 3

7.4 排序并返回序号

>>> rdd = sc.parallelize([1,7,8,5,3,18,34,9,0,12,8]) >>> rdd = rdd.sortBy(lambda x:x).zipWithIndex() >>> pprint(rdd.collect()) [ (0, 0), (1, 1), (3, 2), (5, 3), (7, 4), (8, 5), (8, 6), (9, 7), (12, 8), (18, 9), (34, 10) ] >>> 

7.5 二次排序

翻车，不知道怎么搞，等做完看答案吧

哇，答案给的那个方案太离谱了，还是参考答案好一点

但是参考答案是属于算法吧（摔桌），我一直以为是啥我不知道的API

答案如下，很巧妙：

students = [(“LiLei”,18,87),(“HanMeiMei”,16,77),(“DaChui”,16,66),(“Jim”,18,77),(“RuHua”,18,50)]
rdd_students = sc.parallelize(students)
#rdd_sorted = rdd_students.sortBy(lambda x:*x[2]+x[1],ascending=False)

7.6 连接操作

连接我是会的，但是后面就不太会了

我的思路还一直是API，但是这里用了自定义函数…答案如下：

rdd_classes = sc.parallelize(classes).map(lambda x:(x[1],x[0]))
rdd_scores = sc.parallelize(scores)
rdd_join = rdd_scores.join(rdd_classes).map(lambda t:(t[1][1],t[1][0]))

def average(iterator):
data = list(iterator)
s = 0.0
for x in data:
s = s + x
return s/len(data)

7.7 分组求众数

我能想到用过滤分出不同班级，或者直接group，但是后面不太会，心态小崩。

懂了，又是自定义函数…我的出发点就不对，思维太局限了，答案如下：

#任务：有一批学生信息表格，包括class和age。求每个班级学生年龄的众数。

students = [(“class1”,15),(“class1”,15),(“class2”,16),(“class2”,16),(“class1”,17),(“class2”,19)]
def mode(arr):
dict_cnt = {}
for x in arr:
dict_cnt[x] = dict_cnt.get(x,0)+1
max_cnt = max(dict_cnt.values())
most_values = [k for k,v in dict_cnt.items() if v==max_cnt]
s = 0.0
for x in most_values:
s = s + x
return s/len(most_values)

rdd_students = sc.parallelize(students)
rdd_classes = rdd_students.aggregateByKey([],lambda arr,x:arr+[x],lambda arr1,arr2:arr1+arr2)
rdd_mode = rdd_classes.map(lambda t:(t[0],mode(t[1])))

print(rdd_mode.collect())

---

1. spark持久化操作 persist(),cache() ↩︎
2. pyspark中使用累加器Accumulator统计指标 ↩︎
3. Pyspark累加器(accumulator)陷阱 ↩︎
4. 7道RDD编程练习题 ↩︎

今天的文章 PyS2：RDD编程基础（二）分享到此就结束了，感谢您的阅读。