代码收藏家 Spark中PySpark：添加操作
一、RDD的介绍（了解）
RDD:resilient distributed dataset(弹性分布式数据集合 ) spark的计算核心，spark采用rdd管理数据
RDD
RDD是spark的一种数据模型（规定数据的存储结构和计算方法）
python中的数据模型
list [] 可以重复存储数据 append
set{} 不允许重复存储
dict {k:v} get(key)
RDD的模型可以对内存数进行共享管理
分布式
数据可以在多台服务器上同时计算执行
弹性
可以根据计算的需求将数据进行分区拆分，本质就是将数据分成多份
二、RDD的特点（了解）
分区
可以将计算的海量数据分成多份，需要分成多少可分区可以通过方法指定
每个分区都可以对应一个task线程执行计算
只读
rdd中的数据不能直接修改，需要通过方法计算后得到一个新的rdd
rdd本身存储的数只能读取
依赖
rdd之间是有依赖关系的
新的rdd是通过旧的rdd计算得到
缓存
可以将计算的中结果缓存起来，如果后续计算错误时，可以从缓存位置重新计算
将数据存储在内存或本地磁盘
作用是容错
缓存在执行计算任务程序结束后会释放删除
checkpoint
作用和缓存一样
checkpoint可以将数据存储在分布式存储系统中，比如hdfs
三、创建RDD数据（掌握）
将需要计算的数据转为rdd的数据，就可以利用spark的内存计算方法进行分布式计算操作，这些计算方法就是有rdd提供的
rdd数据的转化方法是有sparkcontext提供的，所以需要先生成sparkcontext，sparkcontext中还包含资源申请和任务划分功能
SparkContext称为Spark的入口类
3-1 Python数据转化为rdd
# 导入sparkcontext
from pyspark import SparkContext
# 创建SparkContext对象
sc = SparkContext()
# 将Python数据转为rdd
# data_int = 10 # 数值类型不能转化rdd
# 能for循环遍历的数据都能转为rdd
data_str = 'abc'
data_list = [1, 2, 3, 4]
data_dict = {'a': 1, 'b': 2}
data_set = {1, 2, 3, 4}
data_tuple = (1, 2, 3, 4)
rdd = sc.parallelize(data_tuple)
# rdd的计算
# rdd的数据输出展示
# 获取所有rdd数据
res = rdd.collect()
print(res)
3-2 文件数据（hdfs）转化为rdd
# 将读取的hdfs文件数据转为rdd
from pyspark import SparkContext
# 生成SparkContext类对象
sc = SparkContext()
# 读取文件数据转为rdd
rdd1 = sc.textFile('hdfs://node1:8020/data')
rdd2 = sc.textFile('/data/words.txt')
# 查看数据
res = rdd1.collect()
print(res)
res = rdd2.collect()
print(res)
3-3 rdd的分区
python数据转发的分区数指定
# RDD分区使用 # 导入sparkcontext from pyspark import SparkContext # 创建SparkContext对象 sc = SparkContext() # 创建生成rdd是可以指定分区数 # Python数据转为rdd指定 # numSlices 可以指定分区数 rdd_py = sc.parallelize([1,2,3,4,5,6],numSlices=10) # rdd计算 # 查看rdd分区数据 res1 = rdd_py.glom().collect() print(res1)
读取的文件数据进行分区数指定
# RDD分区使用
# 导入sparkcontext
from pyspark import SparkContext
# 创建SparkContext对象
sc = SparkContext()
# 创建生成rdd是可以指定分区数
# file文件读取数据指定分区数据
# minPartitions 指定分区
# 文件大小/分区数 = 值 -----余数
# 余数/值 * 100%=百分比 百分比大于10% 会多创建一个分区
rdd_file = sc.textFile('hdfs://node1:8020/data',minPartitions=1)
# 在spark并行度部分会讲解如何根据资源设置分区数
# rdd计算
# 查看rdd分区数据
res2 = rdd_file.glom().collect()
print(res2)
3-4 小文件数据读取
300M 3个块 对应三个分区
在一个目录下，有多个文件，如果文件的大小不够一个块的大小，一个文件就对应一个分区，文件超过一个块，那就一个block（128M）块对应一个分区。
目录下都是小文件，那么读取目录下的文件数据，会对应很多个分区
一个分区对应一个task线程，当小文件过多时，会占用大量的线程，造成资源浪费
使用wholeTextFiles方法可以解决
该方法会现将读取到的数据合并在一起，然后重新进行分区
# 导入sparkcontext
from pyspark import SparkContext
# 创建SparkContext对象
sc = SparkContext(master='yarn')
# rdd = sc.textFile('hdfs://node1:8020/data')
# rdd计算
# wholeTextFiles 会合并小文件数据
# minPartitions 指定分区数
rdd_mini = sc.wholeTextFiles('hdfs://node1:8020/data',minPartitions=1)
# 展示数据
# res1 = rdd.glom().collect()
# print(res1)
res2 = rdd_mini.glom().collect()
print(res2)
java.lang.NoSuchMethodError
java包类的冲突
node1操作 同步到node2和node3
四、常用RDD算子（掌握）
将数据转化为rdd之后，就需要进行rdd的计算了，rdd提供了计算方法
rdd的方法又称为rdd算子
4-1 算子（方法）介绍
rdd中封装了各种算子方便进行计算，主要分为两类
transformation
转化算子 对rdd数据进行转化计算得到新的rdd ，定义了一个线程任务
action
执行算子 触发计算任务，让计算任务进行执行，得到结果
触发线程执行的
rdd的转化算子大部分都是从rdd中读取元素数据（rdd中每条数据），具体计算需要开发人员编写函数传递到rdd算子中
rdd的执行算子则大部分是用来获取数据 collect方法就是触发算子
4-2 常用transformation算子（掌握）
map
rdd.map(lambda 参数:参数计算)
参数接受每个元素数据
# 转化算子map的使用 from pyspark import SparkContext # 创建SparkContext对象 sc = SparkContext() # 生成rdd data = [1, 2, 3, 4] rdd = sc.parallelize(data) # 对rdd进行计算 # 转化算子map使用 # 将处理数据函数当成参数传递给map # 定义函数只需要一个接受参数 def func(x): """ 数据计算逻辑函数 :param x: 接收每一个rdd的元素数据 :return: """ return x + 1 def func2(x): """ 数据计算逻辑函数 :param x: 接收每一个rdd的元素数据 :return: """ 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)
flatMap
处理的是二维嵌套列表数据 [[1,2,3],[4,5,6],[7,8,9]] [1,2,3,4]
rdd.flatMap(lambda 参数:[参数计算])
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)
fliter
rdd.filter(lambda 参数:参数条件过滤)
条件过滤的书写和Python中if判断一样
# 3、过滤算子 rdd7 = sc.parallelize([1, 2, 3, 4]) rdd8 = sc.parallelize(['a', 'b', 'c', 'a']) # filter算子，可以接受rdd中每个元素数据,然后传递给函数进行过滤 # lambda需要有一个接收值x,x接收到每个元素数据后，如何进行过滤需要写判断逻辑 # 判断条件的书写逻辑和if的判断逻辑一样 filter_rdd = rdd7.filter(lambda x: x > 2) filter_rdd2 = rdd8.filter(lambda x: x == 'a')
distinct 去重
不需要lambda rdd.distinct
# 4、去重 # distinct 会对rdd中的重复数据进行去重，去重后会返回一个新的rdd distinct_rdd = rdd8.distinct()
groupBy 分组
rdd.groupBy(lambda 参数:根据参数编写分组条件)
mapValues(list)
# 5、对数据进行分组 # groupBy是分组算子，会读取rdd中每个元素数据，传递给函数使用 # lambda需要一个接收值x，接收groupBy传递的元素数据，然后指定分组规则 # hash(x) % 2 对x中的元素数据进行hash取余，将数据分成两组，余数相同的数据会放在一起 # groupBy返回一个新的rdd,rdd的结构形式是 [(key,value),(k,v)] groupBy_rdd = rdd8.groupBy(lambda x: hash(x) % 2) # 6、对kv形式的数据进行取值处理 # mapValues,可以获取kv中的value值部分传递给函数进行使用 # mapValues返回一个新的rdd数据 mapValues_rdd = groupBy_rdd.mapValues(lambda x: list(x))
k-v数据 [(k,v),(k1,v1)]
groupByKey()
rdd.groupByKey()
reduceByKey()
rdd.reduceByKey(lambda 参数1,参数2:对两个参数计算)
sortByKey()
rdd.sortByKey()
# 6-2 对kv形式的数据进行分组,系统key值得数据会放在一起
rdd9 = sc.parallelize([('a', 1), ('b', 2), ('c', 3), ('a', 1), ('b', 2)])
# 不需要传递处理函数,返回一个新的rdd
groupByKey_rdd = rdd9.groupByKey()
mapValues_rdd2 = groupByKey_rdd.mapValues(lambda x: list(x))
# 6-3 对kv形式的数据先进行分组，在进行聚合计算
# reduceByKey会将相同key的数据放在一起，然后对每个key中对应的valeu进行累加计算
# reduceByKey会将分组后的数据，按照key值传递个函数进行计算
# lambda需要接受两个参数，后面编写累加计算 x=0 y=3 x+y=3 x=3,y=0 res=x+y=3
reduceByKey_rdd = rdd9.reduceByKey(lambda x, y: x + y)
# 6-4 对kv形式的数据先进行排序
# 不需要指定函数,按照key排序，默认升序
sortByKey_rdd= rdd9.sortByKey()
sortByKey_rdd2= rdd9.sortByKey(ascending=False)
sortBy() 排序
rdd.sortBy(lambda x:x,ascending=False)
# 排序算子 # sortBy 可以指定按照哪个数据进行排序 # sortBy会将rdd中的元素数据传递给函数使用 # lambda 需要一个接受值x,接受rdd中每个元素 # 如果元素是kv类型可以通过下标方式指定按照那种排序 x[0] 代表key x[1] 代表value值 # 默认升序 sortBy_rdd = rdd9.sortBy(lambda x: x[1]) # 降序 sortBy_rdd2 = rdd9.sortBy(lambda x: x[1],ascending=False)
4-3 常用action算子（掌握）
collect() 取出rdd中所有值
rdd.collect()
reduce() 非k-v类型数据累加 [1,2,3,4,6]
rdd.reduce(lambda 参数1，参数2:两个参数计算)
count() 统计rdd元素个数
rdd.count()
take() 取出指定数量值
rdd.take(数量)
# 执行算子的使用 from pyspark import SparkContext sc = SparkContext() # python转为rdd rdd = sc.parallelize([1, 2, 3, 4, 5, 6]) # transformation算的转化计算 # map，flatMap，fliter等 # 触发计算 # action算子计算完成返回的是计算结果，不在是rdd了，不能在进行rdd操作了 # collect方法，触发计算获取是所有计算结果 res = rdd.collect() print(res) # reduce方法，传递一个计算逻辑，对元素数据进行累加计算 # 可以不需要转化算直接累加计算，但是不能处理kv形式数据 res = rdd.reduce(lambda x,y:x+y) # x初始为0 y一次获取元素数据 x=0,y=1 x= x+y=1 # x=1,y=2 x+y=3 # x=3,y=3 x+y=6 print(res) # count 获取rdd元素个数 res = rdd.count() print(res) # take取指定数量的元素数据 res=rdd.take(3) print(res)
4-4 词频统计案例（掌握）
# 词频统计
# 导入sparkcontext
from pyspark import SparkContext
# 创建SparkContext对象
sc = SparkContext()
# 将hdfs的文件数据读取后转为rdd
# 第一个参数 指定读取的文件路径
# rdd = sc.textFile('hdfs://node1:8020/data')
# 简写 有的会读取错误,当错误是就写完整
# rdd = sc.textFile('/data')
# 读取某单独文件
rdd = sc.textFile('hdfs://node1:8020/data/words.txt')
# rdd计算
# 对读取到的rdd中的每行数据,先进行切割获取每个单词的数据
# rdd_map = rdd.map(lambda x: x.split(','))
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)
# [('hadoop',1),('flink',1),('spark',2),('hive',2)]
4-5 其他高级算子
多个rdd的方法
union 合并两个rdd 不去重
join k-v类型数据 通过key进行关联
# 多个rdd操作
from pyspark import SparkContext
sc = SparkContext()
rdd1 = sc.parallelize([1,2,3,4])
rdd2 = sc.parallelize([5,6,7,4])
rdd_kv1 = sc.parallelize([('a',1),('b',2),('c',3)])
rdd_kv2 = sc.parallelize([('c',4),('d',5),('e',6)])
# rdd之间的合并
# rdd1和并rdd2,合并后会返回先的rdd
union_rdd = rdd1.union(rdd2)
# rdd3和并rdd4,合并后会返回先的rdd
union_kv_rdd = rdd_kv1.union(rdd_kv2)
# kv形式rdd进行join关联 通过key关联
# 内关联 相同key的数据会保留下来
join_rdd = rdd_kv1.join(rdd_kv2)
# 左关联 左边rdd的数据会被保留下来，如果右边rdd有对应的key值数据会显示，没有对应key值会显示为空
leftOuterJoin_rdd = rdd_kv1.leftOuterJoin(rdd_kv2)
# 右关联 右边rdd的数据会被保留下来，如果左边rdd有对应的key值数据会显示，没有对应key值会显示为空
rightOuterJoin_rdd = rdd_kv1.rightOuterJoin(rdd_kv2)
# 查看结果
res = union_rdd.collect()
print(f'union合并结果:{res}')
res2 = union_kv_rdd.collect()
print(f'kv_union合并结果:{res2}')
res3 = join_rdd.collect()
print(f'join内关联结果:{res3}')
res4 = leftOuterJoin_rdd.collect()
print(f'左关联结果:{res4}')
res5 = rightOuterJoin_rdd.collect()
print(f'右关联结果:{res5}')
重分区
# 1、导入sparkcontext类 from pyspark import SparkContext # 2、初始化SparkContext类型 # 没有指定master参数，默认使用本机资源 sc = SparkContext() # 3、将数据转为rdd数据 # 转化Python数据 data_list = [1, 2, 3, 4, 5, 6] rdd = sc.parallelize(data_list,numSlices=10) map_rdd = rdd.map(lambda x:x+1) # 修改rdd分区信息 # 指定修改的分区数 返回得到新的rdd # repartition 在进行使用时更多进行的是减少分区数 repartition_rdd = map_rdd.repartition(4) repartition_rdd2 = map_rdd.repartition(3) # 查看结果信息 # glom() 查看当前rdd的分区信息 res = map_rdd.glom().collect() print(res) res2 = repartition_rdd.glom().collect() print(res2) res3 = repartition_rdd2.glom().collect() print(res3)
数据保存
# 1、导入sparkcontext类
from pyspark import SparkContext
# 2、初始化SparkContext类型
# 没有指定master参数，默认使用本机资源
sc = SparkContext()
# 3、将数据转为rdd数据
# 转化Python数据
data_list = [1, 2, 3, 4, 5, 6]
rdd = sc.parallelize(data_list,numSlices=3)
rdd.saveAsTextFile('/itcast111')
作者：longjunj
