Spark算子 - Python
目录
第一关 Transformation - map
第二关 Transformation - mapPartitions
第三关 Transformation - filter
第四关 Transformation - flatMap
第五关 Transformation - distinct
第六关 Transformation - sortBy
第七关 Transformation - sortByKey
第八关 Transformation - mapValues
第九关 Transformations - reduceByKey
第十关 Actions - 常用算子
第一关 Transformation - map
from pyspark import SparkContext # -*- coding: UTF-8 -*- 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 **********#第二关 Transformation - mapPartitions
# -*- coding: UTF-8 -*- 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 **********#第三关 Transformation - filter
# -*- 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 **********#第四关 Transformation - flatMap
# -*- 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 **********#第五关 Transformation - distinct
# -*- 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 **********#第六关 Transformation - sortBy
# -*- coding: UTF-8 -*- from pyspark import SparkContext if __name__ == "__main__": # ********** Begin **********# # 1.初始化 SparkContext，该对象是 Spark 程序的入口 sc = SparkContext("local", "Simple App") # 2.创建一个内容为（1, 3, 5, 7, 9, 8, 6, 4, 2）的列表List data = [1, 3, 5, 7, 9, 8, 6, 4, 2] # 3.通过 SparkContext 并行化创建 rdd rdd = sc.parallelize(data) # 4.使用rdd.collect() 收集 rdd 的元素 print(rdd.collect()) """ 使用 sortBy 算子，将 rdd 的数据 (1, 3, 5, 7, 9, 8, 6, 4, 2) 按照下面的规则进行转换操作，规则如下: 需求： 元素排序，例如： 5,4,3,1,2 --> 1,2,3,4,5 """ # 5.使用 sortBy 算子完成以上需求 sort_result = rdd.sortBy(lambda x: x) # 6.使用rdd.collect() 收集完成 sortBy 转换的元素 print(sort_result.collect()) # 7.停止 SparkContext sc.stop() #********** End **********#第七关 Transformation - sortByKey
# -*- coding: UTF-8 -*- from pyspark import SparkContext if __name__ == "__main__": # ********** Begin **********# # 1.初始化 SparkContext，该对象是 Spark 程序的入口 sc = SparkContext("local", "Simple App") # 2.创建一个内容为[(B',1),('A',2),('C',3)]的列表List data = [('B', 1), ('A', 2), ('C', 3)] # 3.通过 SparkContext 并行化创建 rdd rdd = sc.parallelize(data) # 4.使用rdd.collect() 收集 rdd 的元素 print(rdd.collect()) """ 使用 sortByKey 算子，将 rdd 的数据 ('B', 1), ('A', 2), ('C', 3) 按照下面的规则进行转换操作，规则如下: 需求： 元素排序，例如： [(3,3),(2,2),(1,1)] --> [(1,1),(2,2),(3,3)] """ # 5.使用 sortByKey 算子完成以上需求 sort_by_key = rdd.sortByKey() # 6.使用rdd.collect() 收集完成 sortByKey 转换的元素 print(sort_by_key.collect()) # 7.停止 SparkContext sc.stop() # ********** End **********#第八关 Transformation - mapValues
# -*- coding: UTF-8 -*- from pyspark import SparkContext if __name__ == "__main__": # ********** Begin **********# # 1.初始化 SparkContext，该对象是 Spark 程序的入口 sc = SparkContext("local", "Simple App") # 2.创建一个内容为[("1", 1), ("2", 2), ("3", 3), ("4", 4), ("5", 5)]的列表List data = [("1", 1), ("2", 2), ("3", 3), ("4", 4), ("5", 5)] # 3.通过 SparkContext 并行化创建 rdd rdd = sc.parallelize(data) # 4.使用rdd.collect() 收集 rdd 的元素 print(rdd.collect()) """ 使用 mapValues 算子，将 rdd 的数据 ("1", 1), ("2", 2), ("3", 3), ("4", 4), ("5", 5) 按照下面的规则进行转换操作，规则如下: 需求： 元素（key,value）的value进行以下操作： 偶数转换成该数的平方 奇数转换成该数的立方 """ # 5.使用 mapValues 算子完成以上需求 values = rdd.mapValues(lambda x: x * x if x % 2 == 0 else x * x * x) # 6.使用rdd.collect() 收集完成 mapValues 转换的元素 print(values.collect()) # 7.停止 SparkContext sc.stop() # ********** End **********#第九关 Transformations - reduceByKey
# -*- coding: UTF-8 -*- from pyspark import SparkContext if __name__ == "__main__": # ********** Begin **********# # 1.初始化 SparkContext，该对象是 Spark 程序的入口 sc = SparkContext("local", "Simple App") # 2.创建一个内容为[("python", 1), ("scala", 2), ("python", 3), ("python", 4), ("java", 5)]的列表List data = [("python", 1), ("scala", 2), ("python", 3), ("python", 4), ("java", 5)] # 3.通过 SparkContext 并行化创建 rdd rdd = sc.parallelize(data) # 4.使用rdd.collect() 收集 rdd 的元素 print(rdd.collect()) """ 使用 reduceByKey 算子，将 rdd 的数据[("python", 1), ("scala", 2), ("python", 3), ("python", 4), ("java", 5)] 按照下面的规则进行转换操作，规则如下: 需求： 元素（key-value）的value累加操作，例如： (1,1),(1,1),(1,2) --> (1,4) (1,1),(1,1),(2,2),(2,2) --> (1,2),(2,4) """ # 5.使用 reduceByKey 算子完成以上需求 result = rdd.reduceByKey(lambda x, y: x + y) # 6.使用rdd.collect() 收集完成 reduceByKey 转换的元素 print(result.collect()) # 7.停止 SparkContext sc.stop() # ********** End **********#第十关 Actions - 常用算子
# -*- coding: UTF-8 -*- from pyspark import SparkContext if __name__ == "__main__": # ********** Begin **********# # 1.初始化 SparkContext，该对象是 Spark 程序的入口 sc = SparkContext("local", "Simple App") # 2.创建一个内容为[1, 3, 5, 7, 9, 8, 6, 4, 2]的列表List data = [1, 3, 5, 7, 9, 8, 6, 4, 2] # 3.通过 SparkContext 并行化创建 rdd rdd = sc.parallelize(data) # 4.收集rdd的所有元素并print输出 print(rdd.collect()) # 5.统计rdd的元素个数并print输出 print(rdd.count()) # 6.获取rdd的第一个元素并print输出 print(rdd.first()) # 7.获取rdd的前3个元素并print输出 print(rdd.take(3)) # 8.聚合rdd的所有元素并print输出 print(rdd.reduce(lambda x, y: x + y)) # 9.停止 SparkContext sc.stop() # ********** End **********#