图解Spark Transformation算子

布满荆棘的人生 2022-03-19 03:35 327阅读 0赞

## 0. 写在前面 ##

Spark总共有两类算子，分别是**Transformation算子**和**Action算子**。Transformation算子变换不触发提交作业，而Action算子会触发SparkContext提交Job作业，下面主要使用pySpark API来作为事例，图解Spark的Transformation算子。

--------------------

## 1. join ##

![这里写图片描述][20160226210244430]

# joinx = sc.parallelize([('C',4),('B',3),('A',2),('A',1)])y = sc.parallelize([('A',8),('B',7),('A',6),('D',5)])z = x.join(y)print(x.collect())print(y.collect())print(z.collect())[('C', 4), ('B', 3), ('A', 2), ('A', 1)][('A', 8), ('B', 7), ('A', 6), ('D', 5)][('A', (2, 8)), ('A', (2, 6)), ('A', (1, 8)), ('A', (1, 6)), ('B', (3, 7))]
      
       
        1 
        2 
        3 
        4 
        5 
        6 
        7 
        8 
        9 
        10 
        11

--------------------

## 2. leftOuterJoin ##

![这里写图片描述][20160226210329617]

# leftOuterJoinx = sc.parallelize([('C',4),('B',3),('A',2),('A',1)])y = sc.parallelize([('A',8),('B',7),('A',6),('D',5)])z = x.leftOuterJoin(y)print(x.collect())print(y.collect())print(z.collect())[('C', 4), ('B', 3), ('A', 2), ('A', 1)][('A', 8), ('B', 7), ('A', 6), ('D', 5)][('A', (2, 8)), ('A', (2, 6)), ('A', (1, 8)), ('A', (1, 6)), ('C', (4, None)), ('B', (3, 7))]
      
       
        1 
        2 
        3 
        4 
        5 
        6 
        7 
        8 
        9 
        10 
        11

--------------------

## 3. rightOuterJoin ##

![这里写图片描述][20160226210426259]

# rightOuterJoinx = sc.parallelize([('C',4),('B',3),('A',2),('A',1)])y = sc.parallelize([('A',8),('B',7),('A',6),('D',5)])z = x.rightOuterJoin(y)print(x.collect())print(y.collect())print(z.collect())[('C', 4), ('B', 3), ('A', 2), ('A', 1)][('A', 8), ('B', 7), ('A', 6), ('D', 5)][('A', (2, 8)), ('A', (2, 6)), ('A', (1, 8)), ('A', (1, 6)), ('B', (3, 7)), ('D', (None, 5))]
      
       
        1 
        2 
        3 
        4 
        5 
        6 
        7 
        8 
        9 
        10 
        11

--------------------

## 4. partitionBy ##

![这里写图片描述][20160226210546041]

# partitionByx = sc.parallelize([(0,1),(1,2),(2,3)],2)y = x.partitionBy(numPartitions = 3, partitionFunc = lambda x: x)  # only key is passed to paritionFuncprint(x.glom().collect())print(y.glom().collect())[[(0, 1)], [(1, 2), (2, 3)]][[(0, 1)], [(1, 2)], [(2, 3)]]
      
       
        1 
        2 
        3 
        4 
        5 
        6 
        7 
        8

--------------------

## 5. combineByKey ##

![这里写图片描述][20160226210716088]

# combineByKeyx = sc.parallelize([('B',1),('B',2),('A',3),('A',4),('A',5)])createCombiner = (lambda el: [(el,el**2)]) mergeVal = (lambda aggregated, el: aggregated + [(el,el**2)]) # append to aggregatedmergeComb = (lambda agg1,agg2: agg1 + agg2 )  # append agg1 with agg2y = x.combineByKey(createCombiner,mergeVal,mergeComb)print(x.collect())print(y.collect())[('B', 1), ('B', 2), ('A', 3), ('A', 4), ('A', 5)][('A', [(3, 9), (4, 16), (5, 25)]), ('B', [(1, 1), (2, 4)])]
      
       
        1 
        2 
        3 
        4 
        5 
        6 
        7 
        8 
        9 
        10 
        11

--------------------

## 6. aggregateByKey ##

![这里写图片描述][20160226211322372]

# aggregateByKeyx = sc.parallelize([('B',1),('B',2),('A',3),('A',4),('A',5)])zeroValue = [] # empty list is 'zero value' for append operationmergeVal = (lambda aggregated, el: aggregated + [(el,el**2)])mergeComb = (lambda agg1,agg2: agg1 + agg2 )y = x.aggregateByKey(zeroValue,mergeVal,mergeComb)print(x.collect())print(y.collect())[('B', 1), ('B', 2), ('A', 3), ('A', 4), ('A', 5)][('A', [(3, 9), (4, 16), (5, 25)]), ('B', [(1, 1), (2, 4)])]
      
       
        1 
        2 
        3 
        4 
        5 
        6 
        7 
        8 
        9 
        10 
        11

--------------------

## 7. foldByKey ##

![这里写图片描述][20160226211334403]

# foldByKeyx = sc.parallelize([('B',1),('B',2),('A',3),('A',4),('A',5)])zeroValue = 1 # one is 'zero value' for multiplicationy = x.foldByKey(zeroValue,lambda agg,x: agg*x )  # computes cumulative product within each keyprint(x.collect())print(y.collect())[('B', 1), ('B', 2), ('A', 3), ('A', 4), ('A', 5)][('A', 60), ('B', 2)]
      
       
        1 
        2 
        3 
        4 
        5 
        6 
        7 
        8 
        9

--------------------

## 8. groupByKey ##

![这里写图片描述][20160226211345700]

# groupByKeyx = sc.parallelize([('B',5),('B',4),('A',3),('A',2),('A',1)])y = x.groupByKey()print(x.collect())print([(j[0],[i for i in j[1]]) for j in y.collect()])[('B', 5), ('B', 4), ('A', 3), ('A', 2), ('A', 1)][('A', [3, 2, 1]), ('B', [5, 4])]
      
       
        1 
        2 
        3 
        4 
        5 
        6 
        7 
        8

--------------------

## 9. flatMapValues ##

![20160226211357934][]

# flatMapValuesx = sc.parallelize([('A',(1,2,3)),('B',(4,5))])y = x.flatMapValues(lambda x: [i**2 for i in x]) # function is applied to entire value, then result is flattenedprint(x.collect())print(y.collect())[('A', (1, 2, 3)), ('B', (4, 5))][('A', 1), ('A', 4), ('A', 9), ('B', 16), ('B', 25)]
      
       
        1 
        2 
        3 
        4 
        5 
        6 
        7 
        8

--------------------

## 10. mapValues ##

![这里写图片描述][20160226211411825]

# mapValuesx = sc.parallelize([('A',(1,2,3)),('B',(4,5))])y = x.mapValues(lambda x: [i**2 for i in x]) # function is applied to entire valueprint(x.collect())print(y.collect())[('A', (1, 2, 3)), ('B', (4, 5))][('A', [1, 4, 9]), ('B', [16, 25])]
      
       
        1 
        2 
        3 
        4 
        5 
        6 
        7 
        8

--------------------

## 11. groupWith ##

![这里写图片描述][20160226211421294]

# groupWithx = sc.parallelize([('C',4),('B',(3,3)),('A',2),('A',(1,1))])y = sc.parallelize([('B',(7,7)),('A',6),('D',(5,5))])z = sc.parallelize([('D',9),('B',(8,8))])a = x.groupWith(y,z)print(x.collect())print(y.collect())print(z.collect())print("Result:")for key,val in list(a.collect()):     print(key, [list(i) for i in val])[('C', 4), ('B', (3, 3)), ('A', 2), ('A', (1, 1))][('B', (7, 7)), ('A', 6), ('D', (5, 5))][('D', 9), ('B', (8, 8))]Result:D [[], [(5, 5)], [9]]C [[4], [], []]B [[(3, 3)], [(7, 7)], [(8, 8)]]A [[2, (1, 1)], [6], []]
      
       
        1 
        2 
        3 
        4 
        5 
        6 
        7 
        8 
        9 
        10 
        11 
        12 
        13 
        14 
        15 
        16 
        17 
        18 
        19 
        20

--------------------

## 12. cogroup ##

![这里写图片描述][20160226211432569]

# cogroupx = sc.parallelize([('C',4),('B',(3,3)),('A',2),('A',(1,1))])y = sc.parallelize([('A',8),('B',7),('A',6),('D',(5,5))])z = x.cogroup(y)print(x.collect())print(y.collect())for key,val in list(z.collect()):    print(key, [list(i) for i in val])[('C', 4), ('B', (3, 3)), ('A', 2), ('A', (1, 1))][('A', 8), ('B', 7), ('A', 6), ('D', (5, 5))]A [[2, (1, 1)], [8, 6]]C [[4], []]B [[(3, 3)], [7]]D [[], [(5, 5)]]
      
       
        1 
        2 
        3 
        4 
        5 
        6 
        7 
        8 
        9 
        10 
        11 
        12 
        13 
        14 
        15

--------------------

## 13. sampleByKey ##

![这里写图片描述][20160226211443694]

# sampleByKeyx = sc.parallelize([('A',1),('B',2),('C',3),('B',4),('A',5)])y = x.sampleByKey(withReplacement=False, fractions={
        'A':0.5, 'B':1, 'C':0.2})print(x.collect())print(y.collect())[('A', 1), ('B', 2), ('C', 3), ('B', 4), ('A', 5)][('B', 2), ('C', 3), ('B', 4)]
      
       
        1 
        2 
        3 
        4 
        5 
        6 
        7 
        8

--------------------

## 14. subtractByKey ##

![这里写图片描述][20160226211455935]

# subtractByKeyx = sc.parallelize([('C',1),('B',2),('A',3),('A',4)])y = sc.parallelize([('A',5),('D',6),('A',7),('D',8)])z = x.subtractByKey(y)print(x.collect())print(y.collect())print(z.collect())[('C', 1), ('B', 2), ('A', 3), ('A', 4)][('A', 5), ('D', 6), ('A', 7), ('D', 8)][('C', 1), ('B', 2)]
      
       
        1 
        2 
        3 
        4 
        5 
        6 
        7 
        8 
        9 
        10 
        11

--------------------

## 15. subtract ##

![这里写图片描述][20160226211503601]

# subtractx = sc.parallelize([('C',4),('B',3),('A',2),('A',1)])y = sc.parallelize([('C',8),('A',2),('D',1)])z = x.subtract(y)print(x.collect())print(y.collect())print(z.collect())[('C', 4), ('B', 3), ('A', 2), ('A', 1)][('C', 8), ('A', 2), ('D', 1)][('A', 1), ('C', 4), ('B', 3)]
      
       
        1 
        2 
        3 
        4 
        5 
        6 
        7 
        8 
        9 
        10 
        11

--------------------

## 16. keyBy ##

![这里写图片描述][20160226211512554]

# keyByx = sc.parallelize([1,2,3])y = x.keyBy(lambda x: x**2)print(x.collect())print(y.collect())[1, 2, 3][(1, 1), (4, 2), (9, 3)]
      
       
        1 
        2 
        3 
        4 
        5 
        6 
        7 
        8

--------------------

## 17. repartition ##

![这里写图片描述][20160226211520633]

# repartitionx = sc.parallelize([1,2,3,4,5],2)y = x.repartition(numPartitions=3)print(x.glom().collect())print(y.glom().collect())[[1, 2], [3, 4, 5]][[], [1, 2, 3, 4], [5]]
      
       
        1 
        2 
        3 
        4 
        5 
        6 
        7 
        8

--------------------

## 18. coalesce ##

![这里写图片描述][20160226211529966]

# coalescex = sc.parallelize([1,2,3,4,5],2)y = x.coalesce(numPartitions=1)print(x.glom().collect())print(y.glom().collect())[[1, 2], [3, 4, 5]][[1, 2, 3, 4, 5]]
      
       
        1 
        2 
        3 
        4 
        5 
        6 
        7 
        8

--------------------

## 19. zip ##

![这里写图片描述][20160226211537826]

# zipx = sc.parallelize(['B','A','A'])# zip expects x and y to have same #partitions and #elements/partitiony = x.map(lambda x: ord(x))  z = x.zip(y)print(x.collect())print(y.collect())print(z.collect())['B', 'A', 'A'][66, 65, 65][('B', 66), ('A', 65), ('A', 65)]
      
       
        1 
        2 
        3 
        4 
        5 
        6 
        7 
        8 
        9 
        10 
        11 
        12

--------------------

## 20. zipWithIndex ##

![这里写图片描述][20160226211545274]

# zipWithIndexx = sc.parallelize(['B','A','A'],2)y = x.zipWithIndex()print(x.glom().collect())print(y.collect())[['B'], ['A', 'A']][('B', 0), ('A', 1), ('A', 2)]
      
       
        1 
        2 
        3 
        4 
        5 
        6 
        7 
        8

--------------------

## 21. zipWithUniqueId ##

![这里写图片描述][20160226211552466]

# zipWithUniqueIdx = sc.parallelize(['B','A','A'],2)y = x.zipWithUniqueId()print(x.glom().collect())print(y.collect())[['B'], ['A', 'A']][('B', 0), ('A', 1), ('A', 3)]
      
       
        1 
        2 
        3 
        4 
        5 
        6 
        7 
        8

--------------------

【完】

再分享一下我老师大神的人工智能教程吧。零基础！通俗易懂！风趣幽默！还带黄段子！希望你也加入到我们人工智能的队伍中来！[https://blog.csdn.net/jiangjunshow][https_blog.csdn.net_jiangjunshow]

[20160226210244430]: /images/20220319/2040c44858d549fe828ab0aa2cf810ba.png
[20160226210329617]: /images/20220319/2396aa57511d48ee8d957889226611a5.png
[20160226210426259]: /images/20220319/ce75d7bf2ae54dcd9e102f7b0e9845e1.png
[20160226210546041]: /images/20220319/3ea352b1b484465e9d9df393689f3e4c.png
[20160226210716088]: /images/20220319/481fe32547d54a5aa8463f1e55ddf397.png
[20160226211322372]: /images/20220319/2429371b977844028a737aecb401b923.png
[20160226211334403]: /images/20220319/f3303e171c14464b823ec4b83747af2e.png
[20160226211345700]: /images/20220319/f492daeaa8ba4edf9640c439bfc4880d.png
[20160226211357934]: /images/20220319/db65ba1da03a4d39b00083ccfe460822.png
[20160226211411825]: /images/20220319/7e7c117122cc4e9396290f3186e1512a.png
[20160226211421294]: /images/20220319/9ab1e0fd6a444199901bc1eee6ce2e91.png
[20160226211432569]: /images/20220319/57cdd28a40bb4ff387edcd9f7fff034b.png
[20160226211443694]: /images/20220319/90b163e4fc6745358852bb5dad9a2f7a.png
[20160226211455935]: /images/20220319/43464cc27db341a2b20db377ee135391.png
[20160226211503601]: /images/20220319/bdff1d9ef8d84712932e61045e9a6142.png
[20160226211512554]: /images/20220319/8aba5790ebc3484cbf2f7d75e40ccf85.png
[20160226211520633]: /images/20220319/f9c566b9f720498ca38f20c7c30f4fcb.png
[20160226211529966]: /images/20220319/e261e396517d4ad585f7c9efca76c1b8.png
[20160226211537826]: /images/20220319/b54d2c6e85634174abc076dcee8f2504.png
[20160226211545274]: /images/20220319/344f90d3a3cf45d6944f08259050549c.png
[20160226211552466]: /images/20220319/6c0e2362bbd540e2a628cd6e4c44560f.png
[https_blog.csdn.net_jiangjunshow]: https://blog.csdn.net/jiangjunshow/article/details/77338485