Spark算子 - Python （一）_打开右侧代码文件窗口,在 begin 至 end 区域补充代码,完成任务。 在 pyspark 中读

一、理论基础

Spark的算子可分为：

Transformation变换/转换算子：这种变换并不触发提交作业，完成作业中间过程处理。Transformation操作是延迟计算的，也就是说从一个RDD转换生成另一个RDD的操作不是马上执行，需要等到有Action操作的时候才会真正触发运算。

Action行动算子：这类算子会触发SparkContext提交Job作业。Action算子会触发Spark提交作业（Job），并将数据输出Spark系统。
1
2
3

第1关：Transformation - map

任务描述

本关任务：使用Spark的 map 算子按照相关需求完成转换操作。
相关知识

为了完成本关任务，你需要掌握：如何使用map算子。
map

将原来RDD的每个数据项通过map中的用户自定义函数 f 映射转变为一个新的元素。

图中每个方框表示一个RDD 分区，左侧的分区经过自定义函数 f:T->U 映射为右侧的新 RDD 分区。但是，实际只有等到 Action 算子触发后，这个 f 函数才会和其他函数在一个 Stage 中对数据进行运算。
map 案例

    sc = SparkContext("local", "Simple App")
    data = [1,2,3,4,5,6]
    rdd = sc.parallelize(data)
    print(rdd.collect())
    rdd_map = rdd.map(lambda x: x * 2)
    print(rdd_map.collect())
1
2
3
4
5
6

输出：

[1, 2, 3, 4, 5, 6]
[2, 4, 6, 8, 10, 12]

说明：rdd1 的元素（ 1 , 2 , 3 , 4 , 5 , 6 ）经过 map 算子( x -> x*2 )转换成了 rdd2 ( 2 , 4 , 6 , 8 , 10 )。
编程要求

请仔细阅读右侧代码，根据方法内的提示，在Begin - End区域内进行代码补充，具体任务如下：

需求：使用 map 算子，将rdd的数据 (1, 2, 3, 4, 5) 按照下面的规则进行转换操作，规则如下:

偶数转换成该数的平方；
奇数转换成该数的立方。
1
2

测试说明

补充完代码后，点击测评，平台会对你编写的代码进行测试，当你的结果与预期输出一致时，即为通过。

开始你的任务吧，祝你成功！

代码

from pyspark import SparkContext

if __name__ == "__main__":
     #********** Begin **********#
    # 1.初始化 SparkContext，该对象是 Spark 程序的入口
    sc = SparkContext("local", "Simple App")
    # 2.创建一个1到5的列表List
    data = [1, 2, 3, 4, 5]
    # 3.通过 SparkContext 并行化创建 rdd
    rdd = sc.parallelize(data)
    # 4.使用rdd.collect() 收集 rdd 的元素。
    print(rdd.collect())
    """
    使用 map 算子，将 rdd 的数据 (1, 2, 3, 4, 5) 按照下面的规则进行转换操作，规则如下:
    需求：
        偶数转换成该数的平方
        奇数转换成该数的立方
    """
    # 5.使用 map 算子完成以上需求
    rdd_map = rdd.map(lambda x: x * x if x % 2 == 0 else x * x * x)
    # 6.使用rdd.collect() 收集完成 map 转换的元素
    print(rdd_map.collect())
    # 7.停止 SparkContext
    sc.stop()
    #********** End **********#
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

第2关：Transformation - mapPartitions任务描述

本关任务：使用Spark的mapPartitions算子按照相关需求完成转换操作。
相关知识

为了完成本关任务，你需要掌握：如何使用mapPartitions算子。
mapPartitions

mapPartitions函数获取到每个分区的迭代器，在函数中通过这个分区整体的迭 代器对整个分区的元素进行操作。

图中每个方框表示一个RDD分区，左侧的分区经过自定义函数 f:T->U 映射为右侧的新RDD分区。
mapPartitions 与 map

map：遍历算子，可以遍历RDD中每一个元素，遍历的单位是每条记录。

mapPartitions：遍历算子，可以改变RDD格式，会提高RDD并行度，遍历单位是Partition，也就是在遍历之前它会将一个Partition的数据加载到内存中。

那么问题来了，用上面的两个算子遍历一个RDD谁的效率高？
mapPartitions算子效率高。
mapPartitions 案例

def f(iterator):
    list = []
    for x in iterator:
        list.append(x*2)
    return list
if __name__ == "__main__":
    sc = SparkContext("local", "Simple App")
    data = [1,2,3,4,5,6]
    rdd = sc.parallelize(data)
    print(rdd.collect())
    partitions = rdd.mapPartitions(f)
    print(partitions.collect())
1
2
3
4
5
6
7
8
9
10
11
12

输出：

[1, 2, 3, 4, 5, 6]
[2, 4, 6, 8, 10, 12]
1
2

mapPartitions()：传入的参数是rdd的 iterator（元素迭代器），返回也是一个iterator（迭代器）。
编程要求

请仔细阅读右侧代码，根据方法内的提示，在 Begin - End 区域内进行代码补充，具体任务如下：

需求：使用 mapPartitions 算子，将 rdd 的数据 (“dog”, “salmon”, “salmon”, “rat”, “elephant”) 按照下面的规则进行转换操作，规则如下:

将字符串与该字符串的长度组合成一个元组，例如：

    dog  -->  (dog,3)
    salmon   -->  (salmon,6)
1
2
3
4

测试说明

补充完代码后，点击测评，平台会对你编写的代码进行测试，当你的结果与预期输出一致时，即为通过。

开始你的任务吧，祝你成功！

from pyspark import SparkContext

#********** Begin **********#
def f(iterator):
    list = []
    for x in iterator:
        list.append((x, len(x)))
    return list


#********** End **********#

if __name__ == "__main__":
    # 1.初始化 SparkContext，该对象是 Spark 程序的入口
    sc = SparkContext("local", "Simple App")
    # 2. 一个内容为（"dog", "salmon", "salmon", "rat", "elephant"）的列表List
    data = ["dog", "salmon", "salmon", "rat", "elephant"]
    # 3.通过 SparkContext 并行化创建 rdd
    rdd = sc.parallelize(data)
    # 4.使用rdd.collect() 收集 rdd 的元素。
    print(rdd.collect())
    """
    使用 mapPartitions 算子，将 rdd 的数据 ("dog", "salmon", "salmon", "rat", "elephant") 按照下面的规则进行转换操作，规则如下:
    需求：
        将字符串与该字符串的长度组合成一个元组，例如：
        dog  -->  (dog,3)
        salmon   -->  (salmon,6)
    """
    # 5.使用 mapPartitions 算子完成以上需求
    partitions = rdd.mapPartitions(f)
    # 6.使用rdd.collect() 收集完成 mapPartitions 转换的元素
    print(partitions.collect())
    # 7.停止 SparkContext
    sc.stop()

    #********** End **********#
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

第3关：Transformation - filter

100

任务要求
参考答案
评论4

任务描述
相关知识
    filter
    filter 案例
编程要求
测试说明
1
2
3
4
5
6
7
8
9
10

任务描述

本关任务：使用Spark的filter算子按照相关需求完成转换操作。
相关知识

为了完成本关任务，你需要掌握：如何使用filter算子。
filter

filter 函数功能是对元素进行过滤，对每个元素应用f函数，返 回值为 true的元素在RDD中保留，返回值为false的元素将被过滤掉。内部实现相当于生成。

FilteredRDD(this，sc.clean(f))
1

下面代码为函数的本质实现：

   def filter(self, f):
        """
        Return a new RDD containing only the elements that satisfy a predicate.
        >>> rdd = sc.parallelize([1, 2, 3, 4, 5])
        >>> rdd.filter(lambda x: x % 2 == 0).collect()
        [2, 4]
        """
        def func(iterator):
            return filter(fail_on_stopiteration(f), iterator)
        return self.mapPartitions(func, True)
1
2
3
4
5
6
7
8
9
10

上图中每个方框代表一个 RDD 分区， T 可以是任意的类型。通过用户自定义的过滤函数 f，对每个数据项操作，将满足条件、返回结果为 true 的数据项保留。例如，过滤掉 V2 和 V3 保留了 V1，为区分命名为 V’1。
filter 案例

    sc = SparkContext("local", "Simple App")
    data = [1,2,3,4,5,6]
    rdd = sc.parallelize(data)
    print(rdd.collect())
    rdd_filter = rdd.filter(lambda x: x>2)
    print(rdd_filter.collect())
1
2
3
4
5
6

输出：

[1, 2, 3, 4, 5, 6]
[3, 4, 5, 6]
1
2

说明：rdd1( [ 1 , 2 , 3 , 4 , 5 , 6 ] ) 经过 filter 算子转换成 rdd2( [ 3 ,4 , 5 , 6 ] )。
编程要求

请仔细阅读右侧代码，根据方法内的提示，在Begin - End区域内进行代码补充，具体任务如下：

需求1：使用 filter 算子，将 rdd 中的数据 (1, 2, 3, 4, 5, 6, 7, 8) 按照以下规则进行过滤，规则如下:

过滤掉rdd中的所有奇数。
1

测试说明

补充完代码后，点击测评，平台会对你编写的代码进行测试，当你的结果与预期输出一致时，即为通过。

开始你的任务吧，祝你成功！

# -*- coding: UTF-8 -*-
from pyspark import SparkContext

if __name__ == "__main__":
   #********** Begin **********#
    # 1.初始化 SparkContext，该对象是 Spark 程序的入口
    sc = SparkContext("local", "Simple App")
    # 2.创建一个1到8的列表List
    data = [1, 2, 3, 4, 5, 6, 7, 8]
    # 3.通过 SparkContext 并行化创建 rdd
    rdd = sc.parallelize(data)
    # 4.使用rdd.collect() 收集 rdd 的元素。
    print(rdd.collect())
    """
    使用 filter 算子，将 rdd 的数据 (1, 2, 3, 4, 5, 6, 7, 8) 按照下面的规则进行转换操作，规则如下:
    需求：
        过滤掉rdd中的奇数
    """
    # 5.使用 filter 算子完成以上需求
    rdd_filter = rdd.filter(lambda x: x % 2 == 0)
    # 6.使用rdd.collect() 收集完成 filter 转换的元素
    print(rdd_filter.collect())
    # 7.停止 SparkContext
    sc.stop()
    #********** End **********#
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

第4关：Transformation - flatMap

100

任务要求
参考答案
评论4

任务描述
相关知识
    flatMap
    flatMap 案例
编程要求
测试说明
1
2
3
4
5
6
7
8
9
10

任务描述

本关任务：使用Spark的flatMap算子按照相关需求完成转换操作。
相关知识

为了完成本关任务，你需要掌握：如何使用flatMap算子。
flatMap

将原来RDD中的每个元素通过函数f转换为新的元素，并将生成的RDD中每个集合的元素合并为一个集合，内部创建：

FlatMappedRDD(this，sc.clean(f))
1

上图表示RDD的一个分区，进行flatMap函数操作，flatMap中传入的函数为f:T->U，T和U可以是任意的数据类型。将分区中的数据通过用户自定义函数f转换为新的数据。外部大方框可以认为是一个RDD分区，小方框代表一个集合。V1、V2、V3在一个集合作为RDD的一个数据项，可能存储为数组或其他容器，转换为V’1、V’2、V’3后，将原来的数组或容器结合拆散，拆散的数据形成RDD中的数据项。
flatMap 案例

    sc = SparkContext("local", "Simple App")
    data = [["m"], ["a", "n"]]
    rdd = sc.parallelize(data)
    print(rdd.collect())
    flat_map = rdd.flatMap(lambda x: x)
    print(flat_map.collect())
1
2
3
4
5
6

输出：

[['m'], ['a', 'n']]
['m', 'a', 'n']
1
2

flatMap：将两个集合转换成一个集合
编程要求

请仔细阅读右侧代码，根据方法内的提示，在Begin - End区域内进行代码补充，具体任务如下：

需求：使用 flatMap 算子，将rdd的数据 ([1, 2, 3], [4, 5, 6], [7, 8, 9]) 按照下面的规则进行转换操作，规则如下:

合并RDD的元素，例如：

    ([1,2,3],[4,5,6])  -->  (1,2,3,4,5,6)
    ([2,3],[4,5],[6])  -->  (1,2,3,4,5,6)
1
2
3
4

测试说明

补充完代码后，点击测评，平台会对你编写的代码进行测试，当你的结果与预期输出一致时，即为通过。

开始你的任务吧，祝你成功！

# -*- coding: UTF-8 -*-
from pyspark import SparkContext

if __name__ == "__main__":
   	 #********** Begin **********#
       
    # 1.初始化 SparkContext，该对象是 Spark 程序的入口
    sc = SparkContext("local", "Simple App")
 
    # 2.创建一个[[1, 2, 3], [4, 5, 6], [7, 8, 9]] 的列表List
    data = [[1, 2, 3], [4, 5, 6], [7, 8, 9]]
 
    # 3.通过 SparkContext 并行化创建 rdd
    rdd = sc.parallelize(data)
 
    # 4.使用rdd.collect() 收集 rdd 的元素。
    print(rdd.collect())
 
    """
        使用 flatMap 算子，将 rdd 的数据 ([1, 2, 3], [4, 5, 6], [7, 8, 9]) 按照下面的规则进行转换操作，规则如下:
        需求：
            合并RDD的元素，例如：
                            ([1,2,3],[4,5,6])  -->  (1,2,3,4,5,6)
                            ([2,3],[4,5],[6])  -->  (1,2,3,4,5,6)
        """
    # 5.使用 filter 算子完成以上需求
    flat_map = rdd.flatMap(lambda x: x)
 
    # 6.使用rdd.collect() 收集完成 filter 转换的元素
    print(flat_map.collect())
 
    # 7.停止 SparkContext
    sc.stop()
 
    #********** End **********#
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

第5关：Transformation - distinct

100

任务要求
参考答案
评论4

任务描述
相关知识
    distinct
    distinct 案例
编程要求
测试说明
1
2
3
4
5
6
7
8
9
10

任务描述

本关任务：使用 Spark 的 distinct 算子按照需求完成相关操作。
相关知识

为了完成本关任务，你需要掌握：如何使用 distinct 算子。
distinct

distinct 将 RDD 中的元素进行去重操作。

上图中的每个方框代表一个 RDD 分区，通过 distinct 函数，将数据去重。 例如，重复数据 V1、 V1 去重后只保留一份 V1 。
distinct 案例

    sc = SparkContext("local", "Simple App")
    data = ["python", "python", "python", "java", "java"]
    rdd = sc.parallelize(data)
    print(rdd.collect())
    distinct = rdd.distinct()
    print(distinct.collect())
1
2
3
4
5
6

输出：

['python', 'python', 'python', 'java', 'java']
['python', 'java']
1
2

编程要求

请仔细阅读右侧代码，根据方法内的提示，在Begin - End区域内进行代码补充，具体任务如下：

需求：使用 distinct 算子，将 rdd 中的数据进行去重。
测试说明

补充完代码后，点击测评，平台会对你编写的代码进行测试，当你的结果与预期输出一致时，即为通过。

开始你的任务吧，祝你成功！

# -*- coding: UTF-8 -*-
from pyspark import SparkContext

if __name__ == "__main__":
   #********** Begin **********#
    # 1.初始化 SparkContext，该对象是 Spark 程序的入口
    sc = SparkContext("local", "Simple App")
    # 2.创建一个内容为（1, 2, 3, 4, 5, 6, 5, 4, 3, 2, 1）的列表List
    data = [1, 2, 3, 4, 5, 6, 5, 4, 3, 2, 1]
    # 3.通过 SparkContext 并行化创建 rdd
    rdd = sc.parallelize(data)
    # 4.使用rdd.collect() 收集 rdd 的元素
    print(rdd.collect())
    """
       使用 distinct 算子，将 rdd 的数据 (1, 2, 3, 4, 5, 6, 5, 4, 3, 2, 1) 按照下面的规则进行转换操作，规则如下:
       需求：
           元素去重，例如：
                        1,2,3,3,2,1  --> 1,2,3
                        1,1,1,1,     --> 1
       """
    # 5.使用 distinct 算子完成以上需求
    distinctResult = rdd.distinct()
    # 6.使用rdd.collect() 收集完成 distinct 转换的元素
    print(distinctResult.collect())
    # 7.停止 SparkContext
    sc.stop()
    #********** End **********#
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27