Spark算子（RDD）超细致讲解

SPARK算子（RDD）超细致讲解

map,flatmap,sortBykey, reduceBykey,groupBykey,Mapvalues,filter,distinct,sortBy,groupBy共10个转换算子

（一）转换算子

1、map

from pyspark import SparkContext

# 创建SparkContext对象
sc = SparkContext()

# 生成rdd
data = [1, 2, 3, 4]
rdd = sc.parallelize(data)


# 对rdd进行计算
# 转化算子map使用
# 将处理数据函数当成参数传递给map
# 定义函数只需要一个接受参数
def func(x):
    return x + 1


def func2(x):
    return str(x)
# 转化算子执行后会返回新的rdd
rdd_map = rdd.map(func)
rdd_map2 = rdd.map(func2)
rdd_map3 = rdd_map2.map(lambda x: [x])

# 对rdd数据结果展示
# 使用rdd的触发算子,collect获取是所有的rdd元素数据
res = rdd_map.collect()
print(res)

res2 = rdd_map2.collect()
print(res2)


res3 = rdd_map3.collect()
print(res3)

2、flatmap

from pyspark import SparkContext

# 创建SparkContext对象
sc = SparkContext()

# 生成rdd
data = [[1, 2], [3, 4]]
data2 = ['a,b,c','d,f,g']  # 将数据转为['a','b','c','d','f','g']
rdd = sc.parallelize(data)
rdd2 = sc.parallelize(data2)

# rdd计算
# flatMap算子使用  将rdd元素中的列表数依次遍历取出对应的值放入新的rdd [1,2,3,4]
# 传递一个函数,函数接受一个参数
rdd_flatMap = rdd.flatMap(lambda x: x)

rdd_map = rdd2.map(lambda x:x.split(','))
rdd_flatMap2 = rdd_map.flatMap(lambda x:x)

# 输出展示数据
# 使用执行算子
res  = rdd_flatMap.collect()
print(res)

res2  = rdd_map.collect()
print(res2)

res3 = rdd_flatMap2.collect()
print(res3)

3、kv数据结构（sortBykey, reduceBykey,groupBykey,Mapvalues）

# 对kv数据进行处理
from pyspark import SparkContext

# 创建SparkContext对象
sc = SparkContext()
# RDD中的kv结构数据   (k,v)
data = [('a', 1), ('b', 2), ('c', 3), ('a', 1), ('b', 2)]
# 将python数据转化为RDD数据类型
rdd = sc.parallelize(data)

# 排序 根据key排序
# ascending 指定升序还是降序   默认为true是升序  从小到大
rdd_sortByKey = rdd.sortByKey(ascending=False)  # todo 无需传递处理方法

# 根据key聚合计算  将相同key的数据放在一起,然后根据指定的计算方法对分组内的数据进行聚合计算
rdd_reduceByKey = rdd.reduceByKey(lambda x, y: x + y)  # todo 需要传递处理方法

# 分组  只按照key分组 而且无法直接获取到值内容，需要mapValues
rdd_groupByKey = rdd.groupByKey()  # todo 无需传递处理方法

# 获取kv数据中v值
# x接受value值
rdd_mapValues = rdd_groupByKey.mapValues(lambda x: list(x))  # todo 需要传递处理方法

# 展示数据
res = rdd_sortByKey.collect()
print(res)

res2 = rdd_reduceByKey.collect()
print(res2)

res3 = rdd_groupByKey.collect()
print(res3)

res4 = rdd_mapValues.collect()
print(res4)

4、复习（）

data =[hadoop,flink,spark,hive,hive,spark,python,java,python,itcast,itheima] #自己转一下字符串
# rdd计算
# 对读取到的rdd中的每行数据,先进行切割获取每个单词的数据
# rdd_map = rdd.map(lambda x: x.split(','))
#  todo  这段代码给我的感觉就是先进行数据处理再执行flatmap函数。
rdd_flatMap= rdd.flatMap(lambda x: x.split(','))

# 将单词数据转化为k-v结构数据   [(k,v),(k1,v1)]   给每个单词的value一个初始值1
rdd_map_kv = rdd_flatMap.map(lambda x:(x,1))

# 对kv数据进行聚合计算  hive:[1,1]  求和  求平均数  求最大值  求最小值
rdd_reduceByKey =  rdd_map_kv.reduceByKey(lambda x,y:x+y)  # 现将相同key值的数据放在一起,然后对相同key值内的进行累加


# 展示数据
res = rdd.collect()
print(res)

# res2 = rdd_map.collect()
# print(res2)

res3 = rdd_flatMap.collect()
print(res3)

res4 = rdd_map_kv.collect()
print(res4)

res5 = rdd_reduceByKey.collect()
print(res5)

5、过滤算子(filter)

# 数据条件过滤  类似 sql中的where
# 导入sparkcontext
from pyspark import SparkContext

# 创建SparkContext对象
sc = SparkContext()

# 生成rdd数据
data = [1, 2, 3, 4, 5, 6]
rdd = sc.parallelize(data)

# 对rdd数据进行过滤
# filter 接受函数,要求一个函数参数
# lambda x: x > 3   x接受rdd中的每个元素数据    x>3 是过滤条件   过滤条件的数据内容和Python的 if书写内容一样
# 符合条件的数据会放入一个新的rdd
rdd_filter1 = rdd.filter(lambda x: x > 3)
rdd_filter2 = rdd.filter(lambda x: x in [2, 3, 5])
rdd_filter3 = rdd.filter(lambda x: x % 2 == 0)

# 展示数据
res = rdd_filter1.collect()
print(res)

res = rdd_filter2.collect()
print(res)

res = rdd_filter3.collect()
print(res)

6、去重和排序算子（distinct,sortby注意和sortBYkey的区别）

# rdd数据去重
# 导入sparkcontext
from pyspark import SparkContext

# 创建SparkContext对象
sc = SparkContext()

# 生成rdd数据
data = [1, 1, 3, 3, 5, 6]
rdd = sc.parallelize(data)

# 去重
rdd_distinct =  rdd.distinct()

# 获取数据
res =  rdd_distinct.collect()
print(res)

# todo 其它都是传递计算函数，但是这个传递的不是排序函数，而是指定根据谁来排序
# rdd数据排序
# 导入sparkcontext
from pyspark import SparkContext

# 创建SparkContext对象
sc = SparkContext()

# 生成rdd数据
data = [6, 1, 5, 3]
data_kv = [('a',2),('b',3),('c',1)]
rdd = sc.parallelize(data)
rdd_kv = sc.parallelize(data_kv)

# 数据排序
# 传递函数  需要定义一个接收参数
# lambda x:x  第一个参数x 接收rdd中的元素,    第二计算x  指定后会按照该x值排序
# ascending 指定排序规则  默认是True 升序
rdd_sortBy =  rdd.sortBy(lambda x:x,ascending=False)
# 对于kv结构数据  可以通过下标指定按照哪个数据排序
rdd_sortBy2 =  rdd_kv.sortBy(lambda x:x[1],ascending=False)


# 查看结果
res = rdd_sortBy.collect()
print(res)

res2 = rdd_sortBy2.collect()
print(res2)

7、分组算子（groupBy注意和groupBYkey的区别，同时都需要MapValues）

# rdd分组
# 导入sparkcontext
from pyspark import SparkContext

# 创建SparkContext对象
sc = SparkContext()

# 生成rdd数据
data = [6, 1, 5, 3, 10, 22, 35, 17]
rdd = sc.parallelize(data)

# 分组
# 传递函数   函数需要一个接收参数
# lambda x: x % 3   x接收rdd中的每个元素   x%3  对x进行计算,余数相同的数据会放在一起
rdd_groupBy = rdd.groupBy(lambda x: x % 3)
# mapValues 会获取kv中的value数据
# lambda x:list(x)   x接收value值数据
rdd_mapValues = rdd_groupBy.mapValues(lambda x: list(x))

def func(x):
    if len(str(x)) ==1:
        return '1'
    elif  len(str(x)) ==2:
        return 'two'  # 返回分组的key值 指定分组名

rdd_groupBy2 = rdd.groupBy(func)
rdd_mapValues2 = rdd_groupBy2.mapValues(lambda x: list(x))



# 查看结果
res = rdd_groupBy.collect()
print(res)

res2 = rdd_mapValues.collect()
print(res2)

res3 = rdd_groupBy2.collect()
print(res3)

res4 = rdd_mapValues2.collect()
print(res4)