spark的学习-03

news/2024/11/13 6:40:20/

RDD的创建的两种方式：

方式一：并行化一个已存在的集合

方法：parallelize 并行的意思

将一个集合转换为RDD

方式二：读取外部共享存储系统

方法：textFile、wholeTextFile、newAPIHadoopRDD等

读取外部存储系统的数据转换为RDD

RDD的五大特征：

  1. 每个RDD 都由一系列的分区构成

  2. RDD 的转换操作本质上就是对RDD所有分区的并行转换

  3. 每个RDD 都会保存与其他RDD之间的依赖关系：血链机制或者血脉机制

  4. 如果是二元组【KV】类型的RDD，在Shuffle过程中可以自定义分区器，默认是hash分区（hash值取模进行分区）

  5. 可选的，Spark程序运行时，Task的分配可以指定实现本地优先计算：最优计算位置

 
RDD的五大特性分别是什么？ 
 
a. 每个RDD都可以由多个分区构成 
 
b. 对RDD转换处理本质上是对RDD所有分区的并行转换处理 
 
c. 对每个RDD都会保留与其他RDD之间的依赖关系：血脉机制 
 
d. 可选的，对于KV结构的RDD，在经过Shuffle时，可以干预分区规则，默认是Hash分区 
 
e. 可选的，Spark分配Task时会优先本地计算，尽量将Task分配到数据所在的节点
 

 

 

转换算子：
 

map：
 

# map：
list1 = [1, 2, 3, 4, 5]
# 目标是求出集合中各个元素的 3 次方
listRdd = sc.parallelize(list1)
mapRdd = listRdd.map(lambda x: math.pow(x, 3))
mapRdd.foreach(lambda x: print(x))  # foreach是触发算子
 

 

 

flatMap：
 

# flatMap:
# 目标是根据/切割，得到每个歌名
fileRdd = sc.textFile("../../datas/wordcount/songs.txt")
flatMapRdd = fileRdd.flatMap(lambda line: line.split("/"))
flatMapRdd.foreach(lambda x:print(x))
 

 

filter：
 

过滤算子
 

# filter :
# 目标是过滤掉不符合的文本
fileRdd2 = sc.textFile("../../datas/wordcount/songs2.txt")
filterRdd = fileRdd2.filter(lambda line: re.split("\s",line)[2] != '-1' and len(re.split("\s",line)) == 4)
filterRdd.foreach(lambda x: print(x))
 

 

union：
 

联合
 

list2 = [1,2,3,4,5,6,7,8]
list3 = [6,7,8,9,10]
rdd1 = sc.parallelize(list2)
rdd2 = sc.parallelize(list3)rdd3 = rdd1.union(rdd2)rdd3.foreach(lambda x: print(x))   # 1 2 3 4 5 6 7 8 6 7 8 9 10
 

distinct：
 

去重
 

rdd4 = rdd3.distinct()
rdd4.foreach(lambda x: print(x))  # 1 2 3 4 5 6 7 8 9 10
 

分组聚合算子 groupByKey 以及 reduceByKey：
 

groupByKey只根据key进行分组，但不聚合 reduceByKey根据key进行分组，且进行聚合 (必须进行shuffle，可以指定分区的数量和规则) groupByKey转换算子，只对 KV键值对的RDD 起作用
 

rdd5 = sc.parallelize([("word", 10), ("word", 5), ("hello", 100), ("hello", 20), ("laoyan", 1)], numSlices=3)
rdd6 = rdd5.groupByKey()  # ("word",List[10,5])
rdd6.foreach(lambda x: print(x[0], *x[1]))   rdd7 = rdd5.reduceByKey(lambda total, num: total + num)
rdd7.foreach(print)
 

 

重分区算子：repartition、coalesce ：
 

二者都可以将分区变大变小
 

repartition必须经过shuffle 因为底层代码中 shuffle = True，可以将分区变小或者变大
 

而coalesce 可以选择经过不经过shuffle，默认情况下不经过，在默认情况下，只能将分区变小，不能将分区变大。假如shuffle=True，也可以将分区变大。
 

使用repartition更改分区的数量：
 

list01 = [1, 5, 2, 6, 9, 10, 4, 3, 8, 7]
# 没有指定分区，走默认，默认分区个数，因为是local 模式，所以跟核数有关，所以 分区数为2
rdd = sc.parallelize(list01)
print(rdd.getNumPartitions())  # getNumPartitions() 获取分区的数量 返回值不是RDD，所以不是转换算子，是触发算子   # 2
# 使用 repartition 将 分区数量改为4 或 1
changeRdd = rdd.repartition(4)  # 经过shuffle过程，将分区数量更改为4
print(changeRdd.getNumPartitions())  # 现在就将rdd 的分区更改为4了   # 4
# 还可以更改为1 （缩小分区）
print(rdd.repartition(1).getNumPartitions())   # 1
 

使用coalesce 更改分区的数量：
 

将小分区变为大分区，必须进行shuffle过程 在coalesce的中默认shuffle=Flase，所以我们需要手动更改为True
 

changeRdd2 = rdd.coalesce(4,shuffle=True)  #
print(changeRdd2.getNumPartitions())  # 4
 

将大分区改为小分区，在coalesce中可以不进行shuffle过程，所以不需要改为True
 

print(rdd.coalesce(1).getNumPartitions())  # 1 
 

排序算子：sortBy、sortByKey：
 

 fileRdd = sc.textFile("../../datas/c.txt")#fileRdd.sortBy(lambda line:line.split(",")[1],ascending=False).foreach(print)# sortByKey  对KV类型的RDD进行排序rdd5 = sc.parallelize([("word", 10), ("word", 5), ("hello", 100), ("hello", 20), ("laoyan", 1)], numSlices=3)#rdd5.sortByKey(ascending=False).foreach(print)# 假如你想根据value排序，怎么办？rdd5.sortBy(lambda tuple:tuple[1],ascending=False).foreach(print)  # ascending=False降序排序
 

 

触发算子：
 

常见的触发算子：count、foreach、take
 

# 较为常见的触发算子
# count  foreach  saveAsTextFile
# count
list1 = [1,2,3,4,5,6,7,8,9]
rdd1 = sc.parallelize(list1,2)
print(rdd1.count())  #9rdd1.foreach(lambda x: print(x))print(rdd1.take(3))  # [1 2 3]
 

 

其他触发算子：
 

first、take：
 

# first： 返回RDD集合中的第一个元素
print(rdd1.first())  # 1 print(rdd1.take(3))  # [1 2 3]
 

collect：
 

我们在上面sortBy案例中写到了collect，如果不collect就直接打印结果的话，出来的是各个分区中排序的结果，并不是全局的（sortBy是全局排序的，只不过我们之前有分区，只在分区中排序）
 

想看到全局的排序，可以直接将分区数量更改为1，或者直接使用collect收集
 

reduce：
 

我们在上面的案例中也使用到了reduceByKey转换算子，这个和上面的差不多，只不过reduce只进行聚合，而不需要根据key分组什么的，因为就没有key
 

print(rdd1.reduce(lambda sum, num:sum + num))  # 45
 

top 和 takeOrdered：
 

先对RDD中的所有元素进行升序排序，top返回最大的几个元素、takeOrdered返回最小的几个元素
 

都不经过shuffle，将所有元素放入Driver内存中排序，性能更好，只能适合处理小数据量
 

list2 = [2,1,5,79,435,33,576]
rdd2 = sc.parallelize(list2)
print(rdd2.top(3))  # [576, 435, 79]
# takeOrdered 也是一个触发算子，返回排序之后的最小的几个值
print(rdd2.takeOrdered(3))  # [1, 2, 5]
 

join 方面的算子： 
 

join leftOuterJoin rightOuterJoin fullOuterJoin 都为转换算子
 

join的过程，必然引发相同key值的数据汇总在一起，引发shuffle 操作
 

join：
 

rdd_singer_age = sc.parallelize([("周杰伦", 43), ("陈奕迅", 47), ("蔡依林", 41), ("林子祥", 74), ("陈升", 63)],numSlices=2)
rdd_singer_music = sc.parallelize([("周杰伦", "青花瓷"), ("陈奕迅", "孤勇者"), ("蔡依林", "日不落"), ("林子祥", "男儿当自强"),("动力火车", "当")], numSlices=2)# join
joinRdd = rdd_singer_age.join(rdd_singer_music).foreach(lambda x : print(x))   
# ('周杰伦', (43, '青花瓷'))
# ('蔡依林', (41, '日不落'))
# ('陈奕迅', (47, '孤勇者'))
# ('林子祥', (74, '男儿当自强'))
 

leftOuterJoin：
 

和sql中的leftjoin一样，左边的值全出，右边的值有的就显示，没有就显示null
 

rightOuterJoin 同理
 

leftJoinRdd = rdd_singer_age.leftOuterJoin(rdd_singer_music).foreach(lambda x:print(x))
#('周杰伦', (43, '青花瓷'))
#('蔡依林', (41, '日不落'))
#('陈升', (63, None))
#('陈奕迅', (47, '孤勇者'))
#('林子祥', (74, '男儿当自强'))
 

fullOuterJoin：
 

fullJoinRdd = rdd_singer_age.fullOuterJoin(rdd_singer_music).foreach(lambda x: print(x)) 
# ('动力火车', (None, '当'))
# ('周杰伦', (43, '青花瓷'))
# ('蔡依林', (41, '日不落'))
# ('陈升', (63, None))
# ('陈奕迅', (47, '孤勇者'))
# ('林子祥', (74, '男儿当自强'))
 

分区算子 mapPartitions -- 转换算子 foreachParition -- 触发算子
 

mapPartitions：
 

input_rdd = sc.parallelize((1, 2, 3, 4, 5, 6, 7, 8, 9, 10), numSlices=2)
# 使用mapPartitions：对每个分区进行处理
def map_partition(part):rs = [i * 2 for i in part]return rs# 每个分区会调用一次：将这个分区的数据放入内存，性能比map更好，优化型算子，注意更容易出现内存溢出
map_part_rdd = input_rdd.mapPartitions(lambda part: map_partition(part))
 

 

foreachParition：
 

- 优点：性能快、节省外部连接资源 - 缺点：如果单个分区的数据量较大，容易出现内存溢出 - 场景： -数据量不是特别大，需要提高性能【将整个分区的数据放入内存】 -需要构建外部资源时【基于每个分区构建一份资源】
 

def save_part_to_mysql(part):# 构建MySQL连接for i in part:# 利用MySQL连接将结果写入MySQLprint(i)# 将每个分区的数据直接写入MySQL，一个分区就构建一个连接
map_part_rdd.foreachPartition(lambda part: save_part_to_mysql(part))
 

 

 

Spark的容错机制：（重点）
 

1、RDD容错机制：persist持久化机制
 

其中有三个算子： cache 、 persist 、 unpersist
 

cache：
 

# 功能：将RDD缓存在内存中
# 本质其实底层还是调用的 persist ，但是只缓存在内存中，如果内存不足的话，缓存就会失败
语法：cache()
 

persist ：
 

  与cache不同的是，persist 可以自己指定缓存的方式（级别）
 

# 将RDD缓存在磁盘中
StorageLevel.DISK_ONLY = StorageLevel(True, False, False, False)
StorageLevel.DISK_ONLY_2 = StorageLevel(True, False, False, False, 2)
StorageLevel.DISK_ONLY_3 = StorageLevel(True, False, False, False, 3)# 将RDD缓存在内存中
StorageLevel.MEMORY_ONLY = StorageLevel(False, True, False, False)
StorageLevel.MEMORY_ONLY_2 = StorageLevel(False, True, False, False, 2)# 将RDD优先缓存在内存中，如果内存不足，就缓存在磁盘中
StorageLevel.MEMORY_AND_DISK = StorageLevel(True, True, False, False)
StorageLevel.MEMORY_AND_DISK_2 = StorageLevel(True, True, False, False, 2)# 使用堆外内存
StorageLevel.OFF_HEAP = StorageLevel(True, True, True, False, 1)# 使用序列化
StorageLevel.MEMORY_AND_DISK_DESER = StorageLevel(True, True, False, True)
 

常用的有
 

MEMORY_AND_DISK_2   -- 先缓存内存，如果内存不足就缓存在磁盘
MEMORY_AND_DISK_DESER   -- 使用序列化
 

unpersist :
 

功能就是将缓存释放出去
 

  •  
    unpersist(blocking=True)：等释放完再继续下一步 （默认为False）
     
  •  
    场景：明确RDD已经不再使用，后续还有很多的代码需要执行，将RDD的数据从缓存中释放，避免占用资源
     
  •  
    注意：如果不释放，这个Spark程序结束，也会释放这个程序中的所有内存
     
 

总体代码演示：
 

 # step3: 保存结果# 对RDD进行缓存rs_rdd.cache() # 只缓存在内存中rs_rdd.persist(StorageLevel.MEMORY_AND_DISK)# 打印结果：构建RDDrs_rdd.foreach(lambda x: print(x))# 打印第一行：重新构建一遍print(rs_rdd.first())# 统计行数：重新构建一遍print(rs_rdd.count())# todo:3-关闭SparkContexttime.sleep(10000)# 如果这个RDD明确后续代码中不会再被使用，一定要释放缓存rs_rdd.unpersist(blocking=True)# unpersist(blocking=True)：等RDD释放完再继续下一步
# blocking = True：阻塞
 

2、checkPoint 检查点
 

checkpoint需要在触发算子的前面设置检查点，之后设置的话可能会出现只产生文件夹，而不产生结果的情况
 

# 创建sc对象
conf = SparkConf().setMaster("local[2]").setAppName("第一个pysparkDemo")
sc = SparkContext(conf=conf)fileRdd = sc.textFile("../../datas/wordcount/sogou.tsv")
mapRdd = (fileRdd.filter(lambda line: len(re.split("\s+", line)) == 6) \.map(lambda line: (re.split("\s+", line)[0], re.split("\s+", line)[1], re.split("\s+", line)[2][1:-1])))sc.setCheckpointDir("../datas/chk/chk1")mapRdd.checkpoint()
# checkpoint需要在触发算子的前面设置检查点，之后设置的话可能会出现只产生文件夹，而不产生结果的情况print(mapRdd.count())time.sleep(100)sc.stop()
 

 

容错机制面试题：
 

RDD的cache、persist持久化机制和checkpoint检查点机制有什么区别？
 

 

 

  •  
    存储位置
     
    •  
      persist：将RDD缓存在内存或者磁盘中
       
    •  
      chk：将RDD的数据存储在文件系统磁盘中
       
  •  
    生命周期
     
    •  
      persist：当代码中遇到了unpersist或者程序结束，缓存就会被自动清理
       
    •  
      chk：检查点的数据是不会被自动清理的，只能手动删除
       
  •  
    存储内容
     
    •  
      persist：会保留RDD的血脉关系，如果缓存丢失，可以通过血脉进行恢复
       
    •  
      chk：会斩断RDD的血脉关系，不会保留RDD的血脉关系的
       
 

 

 












http://www.ppmy.cn/news/1545749.html












相关文章










【求职】个人介绍









【求职】个人介绍




写这个主要原因想让别人了解你吧，工作跟恋爱一样喜欢才能好好去做，不然都是痛苦。最后可能钱没挣到还有一身病。我也面试很多人，面试最大痛点就是无法短时间内了解对方，面试者自己不说，我如果问不到对方优势&#xff0…




阅读更多...

















Python调用苏信通接口发送短信









Python调用苏信通接口发送短信




毫无征兆的情况下，业务经理通知我，融合云信下架了，当时心里就是一段五味陈杂啊，关键咱也没得选啊，看来接口是要重新开发了。好在业务经理又推荐了苏信通，登录了UI之后，返现功能对比融合云信少了…




阅读更多...

















【go从零单排】泛型(Generics)、链表









【go从零单排】泛型(Generics)、链表




🌈Don’t worry , just coding! 内耗与overthinking只会削弱你的精力，虚度你的光阴，每天迈出一小步，回头时发现已经走了很远。 
📗概念 在Go语言中，泛型（Generics）允许你编写可以处理…




阅读更多...

















WiFi一直获取不到IP地址是怎么回事？









WiFi一直获取不到IP地址是怎么回事？




在当今这个信息化时代，WiFi已成为我们日常生活中不可或缺的一部分。无论是家庭、办公室还是公共场所，WiFi都为我们提供了便捷的无线互联网接入。然而，有时我们可能会遇到WiFi连接后无法获取IP地址的问题，这不仅影响了我们的网络使…




阅读更多...

















UE5.4 PCG 获取地形Layer









UE5.4 PCG 获取地形Layer




使用AttributeFilter：属性过滤器 节点  设置地形Layer名称和权重  效果：




阅读更多...

















基于汇编语言实现的彩色黑白棋游戏









基于汇编语言实现的彩色黑白棋游戏




一、软件背景介绍 
1.1 背景介绍 
黑白棋在西方和日本很流行。游戏通过相互翻转对方的棋子，最后以棋盘上谁的棋子多来判断胜负。它的游戏规则简单，因此上手很容易，但是它的变化又非常复杂。有一种说法是：只需要几分钟学会它，却需要一生的时间去精通它。 
黑白棋是 19 世纪…




阅读更多...

















我自己nodejs练手时常用的一些库基础用法









我自己nodejs练手时常用的一些库基础用法




我自己在使用nodejs以及前端实战练习时常用的一些库的基本使用 
1.bcrypt 
//注册账号时，给密码加密  password是前端传过来的密码，hashPassword是存到数据库中的密码
const bcrypt  require(bcrypt)
const hashPassword  bcrypt.hash(password,10)
//登…




阅读更多...

















解释一下宏定义和函数调用的区别









解释一下宏定义和函数调用的区别




宏定义和函数调用都是用于代码重用和简化编程的工具，但二者之间有许多重要区别。 
定义和作用 宏定义 使用 #define 指令定义，通常用于在预处理阶段进行文本替换。 宏的定义并不进行类型检查，因此可以接受任何类型的参数。 
#define SQUARE(x…




阅读更多...















 
 
 
 










最新文章







 





















Copyright @ 2022~2023