根据文章:
暑假生活每周总结10
from pyspark import SparkContext # 创建SparkContext对象 sc = SparkContext("local", "RDD Transformations Example") # 假设我们有一个RDD数据源 data = sc.parallelize([1, 2, 3, 4, 5]) # 1.map算子 mapped_data = data.map(lambda x: x * 2) print(mapped_data.collect()) # 输出:[2, 4, 6, 8, 10] #2. flatMap算子 nested_data = sc.parallelize([[1, 2], [3, 4]]) flat_mapped_data = nested_data.flatMap(lambda x: x) print(flat_mapped_data.collect()) # 输出:[1, 2, 3, 4] #3. reduceByKey算子,假设是KV型数据 kv_data = sc.parallelize([(1, 'a'), (1, 'b'), (2, 'c')]) reduced_data = kv_data.reduceByKey(lambda a, b: a + b) print(reduced_data.collect()) # 输出:[(1, 'ab'), (2, 'c')] #4. mapValues算子 kv_data2 = sc.parallelize([(1, 10), (2, 20)]) mapped_values_data = kv_data2.mapValues(lambda v: v * 2) print(mapped_values_data.collect()) # 输出:[(1, 20), (2, 40)] #5. groupBy算子 grouped_data = data.groupBy(lambda x: x % 2) print(grouped_data.collect()) # 输出:[(0, <pyspark.resultiterable.ResultIterable object at ...>), (1, <pyspark.resultiterable.ResultIterable object at ...>)] #6. filter算子 filtered_data = data.filter(lambda x: x % 2 == 0) print(filtered_data.collect()) # 输出:[2, 4] # 7.distinct算子 distinct_data = data.distinct() print(distinct_data.collect()) # 输出:[1, 2, 3, 4, 5] #8. union算子 rdd1 = sc.parallelize([1, 2]) rdd2 = sc.parallelize([3, 4]) union_data = rdd1.union(rdd2) print(union_data.collect()) # 输出:[1, 2, 3, 4] #9. join算子 rdd_kv1 = sc.parallelize([(1, 'a'), (2, 'b')]) rdd_kv2 = sc.parallelize([(1, 'x'), (2, 'y')]) joined_data = rdd_kv1.join(rdd_kv2) print(joined_data.collect()) # 输出:[(1, ('a', 'x')), (2, ('b', 'y'))]
# 10. intersection算子
rdd1 = sc.parallelize([1, 2, 3, 4, 5])
rdd2 = sc.parallelize([4, 5, 6, 7, 8])
intersection_data = rdd1.intersection(rdd2)
print(intersection_data.collect()) # 输出:[4, 5]
# 11. glom算子
data = sc.parallelize(range(10), numSlices=4) # 创建一个分区数为4的RDD
glom_data = data.glom()
for part in glom_data.collect():
print(part) # 输出每个分区内的元素列表
# 12. groupByKey算子
kv_data = sc.parallelize([(1, 'a'), (1, 'b'), (2, 'c'), (3, 'd')])
grouped_data = kv_data.groupByKey()
for key, values in grouped_data.collect():
print(f"Key: {key}, Values: {list(values)}") # 输出:Key: 1, Values: ['a', 'b'], Key: 2, Values: ['c'], Key: 3, Values: ['d']
#13. sortBy算子 sorted_data = data.sortBy(lambda x: -x) # 降序排序 print(sorted_data.collect()) # 输出:[5, 4, 3, 2, 1] #14. sortByKey算子(针对KV型) kv_sorted_data = sc.parallelize([(2, 'b'), (1, 'a'), (3, 'c')]) sorted_by_key = kv_sorted_data.sortByKey(ascending=True) print(sorted_by_key.collect()) # 输出:[(1, 'a'), (2, 'b'), (3, 'c')]
1.map算子,将rdd数据进行依次处理,基于map的算子中所收集的逻辑函数。进行逻辑运算,返回rdd,比较常见。
2.flatMap算子,对rdd执行map操作,进行解嵌套,返回rdd。
3.reduceByKey算子,主要是针对KV型数据进行key分组进行数据的计算,根据所提供的逻辑,进行组内数据的聚合操作。
4.mapValues算子,主要是针对二元元组的rdd,对其内部的value执行map操作,传入的逻辑函数,主要是对于value值得操作。
5.groupBy算子,对rdd的数据进行分组。分组完成后返回二元元组,同组数据放入一个迭代器中作为value值。
6.filter算子,主要是过滤数据。传入的函数要有bool类型的返回值。
7.distinct算子,主要是进行数据的去重操作,返回新的rdd。
8.union算子,主要是对于连个rdd,进行合并返回。
9.join算子,对于两个rdd可以执行join操作,相当于sql中的内外连接。语法有join,leftOuterJoin,rightOuterJoin。
10.intersection算子,主要是求取两个rdd的交集,并返回新的rdd。
11.glom算子,主要是进行对于rdd按照其分区进行嵌套。
12.groupByKey算子,针对KV型rdd,自然按照key进行分组。
13.sortBy算子,主要是对rdd数据进行排序,基于你指定的排序依据。sortBy(func,ascending=False,numPartitions=1) true为升序,false为降序。
14.sortByKey算子,主要是针对KV型rdd,按照key进行排序。sortByKey(ascending=True,numPartitions=1,keyfunc=所要书写的处理方法,相当于在排序前的数据处理)实例化python代码实现
标签:常用,rddTransformation,sc,collect,rdd,算子,print,data From: https://www.cnblogs.com/syhxx/p/17995065