Spark (Python版) 零基础学习笔记（二）—— Spark Transformations总结及举例

1. map(func) 
将func函数作用到数据集的每个元素，生成一个新的分布式的数据集并返回

>>> a = sc.parallelize(('a', 'b', 'c'))
>>> a.map(lambda x: x+'1').collect()
['a1', 'b1', 'c1']

2. filter(func) 
选出所有func返回值为true的元素，作为一个新的数据集返回

>>> a = sc.parallelize(range(10))
>>> a.filter(lambda x: x%2==0).collect()  # 选出0-9的偶数
[0, 2, 4, 6, 8]

3. flatMap(func) 
与map相似，但是每个输入的item能够被map到0个或者更多的items输出，也就是说func的返回值应当是一个Sequence，而不是一个单独的item

>>> l = ['I am Tom', 'She is Jenny', 'He is Ben']
>>> a = sc.parallelize(l,3)
>>> a.flatMap(lambda line: line.split()).collect()  # 将每个字符串中的单词划分出来
['I', 'am', 'Tom', 'She', 'is', 'Jenny', 'He', 'is', 'Ben']

4. mapPartitions(func) 
与map相似，但是mapPartitions的输入函数单独作用于RDD的每个分区(block)上，因此func的输入和返回值都必须是迭代器iterator。 
例如：假设RDD有十个元素0~9，分成三个区，使用mapPartitions返回每个元素的平方。如果使用map方法，map中的输入函数会被调用10次，而使用mapPartitions方法，输入函数只会被调用3次，每个分区被调用1次。

>>> def squareFunc(a):
. . .     for i in a:
. . .         yield i*i
. . .
>>> a = sc.parallelize(range(10), 3)
PythonRDD[1] at RDD at PythonRDD.scala:48
>>> a.mapPartitions(squareFunc).collect()
[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

5. mapPartitionsWithIndex(func) 
与mapPartitions相似，但是输入函数func提供了一个正式的参数，可以用来表示分区的编号。

>>> def func(index, iterator):  # 返回每个分区的编号和数值
. . .     yield (‘index ‘ + str(index) + ’ is: ‘ + str(list(iterator)))
. . .
>>> a = sc.parallelize(range(10),3)
>>> a.mapPartitionsWithIndex(func).collect()
['index 0 is: [0, 1, 2]', 'index 1 is: [3, 4, 5]', 'index 2 is: [6, 7, 8, 9]']
>>> def squareIndex(index, iterator):  # 返回每个数值所属分区的编号和数值的平方
...     for i in iterator:
...         yield ("The index is: " + str(index) + ", and the square is: " + str(i*i))
... 
>>> a.mapPartitionsWithIndex(squareIndex).collect()
['The index is: 0, and the square is: 0', 
'The index is: 0, and the square is: 1', 
'The index is: 1, and the square is: 4', 
'The index is: 1, and the square is: 9', 
'The index is: 1, and the square is: 16', 
'The index is: 2, and the square is: 25', 
'The index is: 2, and the square is: 36', 
'The index is: 3, and the square is: 49', 
'The index is: 3, and the square is: 64', 
'The index is: 3, and the square is: 81']

6. sample(withReplacement, fraction, seed) 
从数据中抽样，withReplacement表示是否有放回，withReplacement=true表示有放回抽样，fraction为抽样的概率（0<=fraction<=1），seed为随机种子。 
例如：从1-100之间抽取样本，被抽取为样本的概率为0.2

>>> data = sc.parallelize(range(1,101),2)
>>> sample = data.sample(True, 0.2)
>>> sampleData.count()
19
>>> sampleData.collect()
[16, 19, 24, 29, 32, 33, 44, 45, 55, 56, 56, 57, 65, 65, 73, 83, 84, 92, 96]

！！！注意，Spark中的sample抽样，当withReplacement=True时，相当于采用的是泊松抽样；当withReplacement=False时，相当于采用伯努利抽样，fraction并不是表示抽样得到的样本占原来数据总量的百分比，而是一个元素被抽取为样本的概率。fraction=0.2并不是说明要抽出100个数字中20%的数据作为样本，而是每个数字被抽取为样本的概率为0.2，这些数字被认为来自同一总体，样本的大小并不是固定的，而是服从二项分布。

7. union(otherDataset) 
并集操作，将源数据集与union中的输入数据集取并集，默认保留重复元素（如果不保留重复元素，可以利用distinct操作去除，下边介绍distinct时会介绍）。

>>> data1 = sc.parallelize(range(10))
>>> data2 = sc.parallelize(range(6,15))
>>> data1.union(data2).collect()
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 6, 7, 8, 9, 10, 11, 12, 13, 14]

8. intersection(otherDataset) 
交集操作，将源数据集与union中的输入数据集取交集，并返回新的数据集。

>>> data1 = sc.parallelize(range(10))
>>> data2 = sc.parallelize(range(6,15))
>>> data1.intersection(data2).collect()
[8, 9, 6, 7]

9. distinct([numTasks]) 
去除数据集中的重复元素。

>>> data1 = sc.parallelize(range(10))
>>> data2 = sc.parallelize(range(6,15))
>>> data1.union(data2).distinct().collect()
[0, 8, 1, 9, 2, 10, 11, 3, 12, 4, 5, 13, 14, 6, 7]

下边的一系列transactions会用的键（Key）这一概念，在进行下列有关Key操作时使用的数据集为记录伦敦各个片区（英文称为ward）中学校和学生人数相关信息的表格，下载地址： 
https://data.london.gov.uk/dataset/london-schools-atlas/resource/64f771ee-38b1-4eff-8cd2-e9ba31b90685# 
下载后将其中命名为WardtoSecSchool_LDS_2015的sheet里边的数据保存为csv格式，删除第一行的表头，并重新命名为school.csv 
数据格式为： 
(Ward_CODE, Ward_NAME, TotalWardPupils, Ward2Sec_Flow_No., Secondary_School_URN, Secondary_School_Name, Pupil_count) 
首先对数据进行一些预处理:

>>> school = sc.textFile("file:///home/yang/下载/school.csv")  
Data = sc.textFile("file:///home/yang/下载/school.csv") 
>>> school.count()  # 共有16796行数据
16796
>>> import re  # 引入python的正则表达式包
>>> rows = school.map(lambda line: re.subn(',[s]+',': ', line))

注意：1. 从本地读取数据时，代码中要通过 “file://” 前缀指定读取本地文件。Spark shell 默认是读取 HDFS 中的文件，需要先上传文件到 HDFS 中，否则会有“org.apache.hadoop.mapred.InvalidInputException: Input path does not exist: hdfs://localhost:9000/user/hadoop/school.csv”的错误。 
2. 对数据集进行了一下预处理，利用正则匹配替换字符串，由于一些学校的名字的字符串中本身含有逗号，比如“The City Academy, Hackney”， 此时如果利用csv的分隔符’,’进行分割，并不能将名字分割为“The City Academy”和“Hackney”。我们注意到csv的分隔符逗号后边是没有空格的，而名字里边的逗号后边都会有空格（英语书写习惯），因此，先利用re.subn语句对逗号后边含有至少一个空格（正则表达式为’,[s]+’）的子字符串进行替换，替换为’: ’，然后再进行后续操作。以上即为对这一数据集的预处理过程。

10. groupByKey([numTasks]) 
作用于由键值对(K, V)组成的数据集上，将Key相同的数据放在一起，返回一个由键值对(K, Iterable)组成的数据集。 
注意：1. 如果这一操作是为了后续在每个键上进行聚集（aggregation），比如sum或者average，此时使用reduceByKey或者aggregateByKey的效率更高。2. 默认情况下，输出的并行程度取决于RDD分区的数量，但也可以通过给可选参数numTasks赋值来调整并发任务的数量。

>>> newRows = rows.map(lambda r: r[0].split(','))  
>>> ward_schoolname = newRows .map(lambda r: (r[1], r[5])).groupByKey()  # r[1]为ward的名字，r[5]为学校的名字
>>> ward_schoolname.map(lambda x: {x[0]: list(x[1])}).collect()  # 列出每个ward区域内所有的学校的名字
[{'Stifford Clays': ['William Edwards School', 'Brentwood County High School', "The Coopers' Company and Coborn School", 'Becket Keys Church of England Free School', ...] 
# 输出结果为在Stifford Clays这个ward里的学校有William Edwards School，Brentwood County High School，The Coopers' Company and Coborn School等等...

11. reduceByKey(func, [numTasks]) 
作用于键值对(K, V)上，按Key分组，然后将Key相同的键值对的Value都执行func操作，得到一个值，注意func的类型必须满足

>>> pupils = newRows.map(lambda r: (r[1], int(r[6])))  # r[1]为ward的名字，r[6]为每个学校的学生数
>>> ward_pupils = pupils.reduceByKey(lambda x, y: x+y)   # 计算各个ward中的学生数
>>> ward_pupils.collect()  # 输出各个ward中的学生数
[('Stifford Clays', 1566), ('Shenley', 1625), ('Southbury', 3526), 
('Rainham and Wennington', 769), ('Bromley Town', 574), ('Waltham Abbey Honey Lane', 835), 
('Telegraph Hill', 1238), ('Chigwell Village', 1506), ('Gooshays', 2097), ('Edgware', 2585), 
('Camberwell Green', 1374), ('Glyndon', 4633),...]

12. aggregateByKey(zeroValue, seqOp, comOp, [numTasks]) 
在于键值对(K, V)的RDD中，按key将value进行分组合并，合并时，将每个value和初始值作为seqOp函数的参数，进行计算，返回的结果作为一个新的键值对(K, V)，然后再将结果按照key进行合并，最后将每个分组的value传递给comOp函数进行计算（先将前两个value进行计算，将返回结果和下一个value传给comOp函数，以此类推），将key与计算结果作为一个新的键值对(K, V)输出。 
例子： 上述统计ward内学生人数的操作也可以通过aggregateByKey实现，此时，seqOp和comOp都是进行加法操作，代码如下：

>>> ward_pupils = pupils.aggregateByKey(0, lambda x, y: x+y, lambda x, y: x+y)
>>> ward_pupils.collect()  
[('Stifford Clays', 1566), ('Shenley', 1625), ('Southbury', 3526), 
('Rainham and Wennington', 769), ('Bromley Town', 574), ('Waltham Abbey Honey Lane', 835), 
('Telegraph Hill', 1238), ('Chigwell Village', 1506), ('Gooshays', 2097), ('Edgware', 2585), 
('Camberwell Green', 1374), ('Glyndon', 4633),...]

13. sortByKey([ascending=True], [numTasks]) 
按照Key进行排序，ascending的值默认为True，True/False表示升序还是降序 
例如：将上述ward按照ward名字降序排列，打印出前十个

>>> ward_pupils.sortByKey(False, 4).take(10)
[('Yiewsley', 2560), ('Wormholt and White City', 1455), ('Woodside', 1204), 
('Woodhouse', 2930), ('Woodcote', 1214), ('Winchmore Hill', 1116), ('Wilmington', 2243), 
('Willesden Green', 1896), ('Whitefoot', 676), ('Whalebone', 2294)]

14. join(otherDataset, [numTasks]) 
类似于SQL中的连接操作，即作用于键值对(K, V)和(K, W)上，返回元组 (K, (V, W))，spark也支持外连接，包括leftOuterJoin，rightOuterJoin和fullOuterJoin。例子：

>>> class1 = sc.parallelize(('Tom', 'Jenny', 'Bob')).map(lambda a: (a, 'attended'))
>>> class2 = sc.parallelize(('Tom', 'Amy', 'Alice', 'John')).map(lambda a: (a, 'attended'))
>>> class1.join(class2).collect()
[('Tom', ('attended', 'attended'))]
>>> class1.leftOuterJoin(class2).collect()
[('Tom', ('attended', 'attended')), ('Jenny', ('attended', None)), ('Bob', ('attended', None))]
>>> class1.rightOuterJoin(class2).collect()
[('John', (None, 'attended')), ('Tom', ('attended', 'attended')), ('Amy', (None, 'attended')), ('Alice', (None, 'attended'))]
>>> class1.fullOuterJoin(class2).collect()
[('John', (None, 'attended')), ('Tom', ('attended', 'attended')), ('Jenny', ('attended', None)), ('Bob', ('attended', None)), ('Amy', (None, 'attended')), ('Alice', (None, 'attended'))]

15. cogroup(otherDataset, [numTasks]) 
作用于键值对(K, V)和(K, W)上，返回元组 (K, (Iterable, Iterable))。这一操作可叫做groupWith。

>>> class1 = sc.parallelize(('Tom', 'Jenny', 'Bob')).map(lambda a: (a, 'attended'))
>>> class2 = sc.parallelize(('Tom', 'Amy', 'Alice', 'John')).map(lambda a: (a, 'attended'))
>>> group = class1.cogroup(class2)
>>> group.collect()
[('John', (<pyspark.resultiterable.ResultIterable object at 0x7fb7e808afd0>, <pyspark.resultiterable.ResultIterable object at 0x7fb7e808a1d0>)), 
('Tom', (<pyspark.resultiterable.ResultIterable object at 0x7fb7e808a7f0>, <pyspark.resultiterable.ResultIterable object at 0x7fb7e808a048>)), 
('Jenny', (<pyspark.resultiterable.ResultIterable object at 0x7fb7e808a9b0>, <pyspark.resultiterable.ResultIterable object at 0x7fb7e808a208>)), 
('Bob', (<pyspark.resultiterable.ResultIterable object at 0x7fb7e808ae80>, <pyspark.resultiterable.ResultIterable object at 0x7fb7e8b448d0>)), 
('Amy', (<pyspark.resultiterable.ResultIterable object at 0x7fb7e8b44c88>, <pyspark.resultiterable.ResultIterable object at 0x7fb7e8b44588>)), 
('Alice', (<pyspark.resultiterable.ResultIterable object at 0x7fb7e8b44748>, <pyspark.resultiterable.ResultIterable object at 0x7fb7e8b44f98>))]
>>> group.map(lambda x: {x[0]: [list(x[1][0]), list(x[1][1])]}).collect()
[{'John': [[], ['attended']]}, {'Tom': [['attended'], ['attended']]}, {'Jenny': [['attended'], []]}, {'Bob': [['attended'], []]}, {'Amy': [[], ['attended']]}, {'Alice': [[], ['attended']]}]

16. cartesian(otherDataset) 
笛卡尔乘积，作用于数据集T和U上，返回(T, U)，即数据集中每个元素的两两组合

>>> a = sc.parallelize(('a', 'b', 'c'))
>>> b = sc.parallelize(('d', 'e', 'f'))
>>> a.cartesian(b).collect()
[('a', 'd'), ('a', 'e'), ('a', 'f'), ('b', 'd'), ('b', 'e'), ('b', 'f'), ('c', 'd'), ('c', 'e'), ('c', 'f')]

17. pipe(command, [envVars]) 
将驱动程序中的RDD交给shell处理（外部进程），例如Perl或bash脚本。RDD元素作为标准输入传给脚本，脚本处理之后的标准输出会作为新的RDD返回给驱动程序。

18. coalesce(numPartitions) 
将RDD的分区数减小到numPartitions个。当数据集通过过滤规模减小时，使用这个操作可以提升性能。

19. repartition(numPartitions) 
重组数据，数据被重新随机分区为numPartitions个，numPartitions可以比原来大，也可以比原来小，平衡各个分区。这一操作会将整个数据集在网络中重新洗牌。

20. repartitionAndSortWithinPartitions(partitioner) 
根据给定的partitioner函数重新将RDD分区，并在分区内排序。这比先repartition然后在分区内sort高效，原因是这样迫使排序操作被移到了shuffle阶段。