Spark算子 - Python

第1关：Transformation - map

# -*- coding: UTF-8 -*-
from pyspark import SparkContextif __name__ == "__main__":#********** Begin **********## 1.初始化 SparkContext，该对象是 Spark 程序的入口sc = SparkContext("local","Simple App")# 2.创建一个1到5的列表ListList = [1,2,3,4,5]# 3.通过 SparkContext 并行化创建 rddrdd = sc.parallelize(List)# 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.停止 SparkContextsc.stop()#********** End **********#

第2关：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__":#********** Begin **********## 1.初始化 SparkContext，该对象是 Spark 程序的入口sc = SparkContext("local","Simple App")# 2. 一个内容为（"dog", "salmon", "salmon", "rat", "elephant"）的列表Listdata = ["dog", "salmon", "salmon", "rat", "elephant"]# 3.通过 SparkContext 并行化创建 rddrdd = 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.停止 SparkContextsc.stop()#********** End **********#

第3关：Transformation - filter

# -*- coding: UTF-8 -*-
from pyspark import SparkContextif __name__ == "__main__":#********** Begin **********## 1.初始化 SparkContext，该对象是 Spark 程序的入口sc = SparkContext("local","Simple App")# 2.创建一个1到8的列表Listdata = [1,2,3,4,5,6,7,8]# 3.通过 SparkContext 并行化创建 rddrdd = 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.停止 SparkContextsc.stop()#********** End **********#

第4关：Transformation - flatMap

# -*- coding: UTF-8 -*-
from pyspark import SparkContextif __name__ == "__main__":#********** Begin **********## 1.初始化 SparkContext，该对象是 Spark 程序的入口sc = SparkContext("local","Simple App")# 2.创建一个[[1, 2, 3], [4, 5, 6], [7, 8, 9]] 的列表Listdata = [[1,2,3],[4,5,6],[7,8,9]]# 3.通过 SparkContext 并行化创建 rddrdd = 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.停止 SparkContextsc.stop()#********** End **********#

第5关：Transformation - distinct

# -*- coding: UTF-8 -*-
from pyspark import SparkContextif __name__ == "__main__":#********** Begin **********## 1.初始化 SparkContext，该对象是 Spark 程序的入口sc = SparkContext("local","Simple App")# 2.创建一个内容为（1, 2, 3, 4, 5, 6, 5, 4, 3, 2, 1）的列表Listdata = [1,2,3,4,5,6,5,4,3,2,1]# 3.通过 SparkContext 并行化创建 rddrdd = 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,31,1,1,1,     --> 1"""# 5.使用 distinct 算子完成以上需求a = rdd.distinct()# 6.使用rdd.collect() 收集完成 distinct 转换的元素print(a.collect())# 7.停止 SparkContextsc.stop()#********** End **********#

第6关：Transformation - sortBy

# -*- coding: UTF-8 -*-
from pyspark import SparkContextif __name__ == "__main__":# ********** Begin **********## 1.初始化 SparkContext，该对象是 Spark 程序的入口sc = SparkContext("local", "Simple App")# 2.创建一个内容为（1, 3, 5, 7, 9, 8, 6, 4, 2）的列表Listdata = [1,3,5,7,9,8,6,4,2]# 3.通过 SparkContext 并行化创建 rddrdd = 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 算子完成以上需求a = rdd.sortBy(lambda x:x)# 6.使用rdd.collect() 收集完成 sortBy 转换的元素print(a.collect())# 7.停止 SparkContextsc.stop()#********** End **********#

第7关：Transformation - sortByKey

# -*- coding: UTF-8 -*-
from pyspark import SparkContextif __name__ == "__main__":# ********** Begin **********## 1.初始化 SparkContext，该对象是 Spark 程序的入口sc = SparkContext("local", "Simple App")# 2.创建一个内容为[(B',1),('A',2),('C',3)]的列表Listdata = [("B",1),("A",2),("C",3)]# 3.通过 SparkContext 并行化创建 rddrdd = 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 算子完成以上需求a = rdd.sortByKey()# 6.使用rdd.collect() 收集完成 sortByKey 转换的元素print(a.collect())# 7.停止 SparkContextsc.stop()# ********** End **********#

第8关：Transformation - mapValues

# -*- coding: UTF-8 -*-
from pyspark import SparkContextif __name__ == "__main__":# ********** Begin **********## 1.初始化 SparkContext，该对象是 Spark 程序的入口sc = SparkContext("local", "Simple App")# 2.创建一个内容为[("1", 1), ("2", 2), ("3", 3), ("4", 4), ("5", 5)]的列表Listdata = [("1",1),("2",2),("3",3),("4",4),("5",5)]# 3.通过 SparkContext 并行化创建 rddrdd = 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 算子完成以上需求a = rdd.mapValues(lambda x:x*x if x%2==0 else x*x*x)# 6.使用rdd.collect() 收集完成 mapValues 转换的元素print(a.collect())# 7.停止 SparkContextsc.stop()# ********** End **********#

第9关：Transformations - reduceByKey

# -*- coding: UTF-8 -*-
from pyspark import SparkContextif __name__ == "__main__":# ********** Begin **********## 1.初始化 SparkContext，该对象是 Spark 程序的入口sc = SparkContext("local", "Simple App")# 2.创建一个内容为[("python", 1), ("scala", 2), ("python", 3), ("python", 4), ("java", 5)]的列表Listdata = [("python", 1), ("scala", 2), ("python", 3), ("python", 4), ("java", 5)]# 3.通过 SparkContext 并行化创建 rddrdd = 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 算子完成以上需求a = rdd.reduceByKey(lambda x,y:x+y)# 6.使用rdd.collect() 收集完成 reduceByKey 转换的元素print(a.collect())# 7.停止 SparkContextsc.stop()# ********** End **********#

第10关：Actions - 常用算子

# -*- coding: UTF-8 -*-
from pyspark import SparkContextif __name__ == "__main__":# ********** Begin **********## 1.初始化 SparkContext，该对象是 Spark 程序的入口sc = SparkContext("local","Simple App")# 2.创建一个内容为[1, 3, 5, 7, 9, 8, 6, 4, 2]的列表Listdata = [1, 3, 5, 7, 9, 8, 6, 4, 2]# 3.通过 SparkContext 并行化创建 rddrdd = 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.停止 SparkContextsc.stop()# ********** End **********#