首页 > 其他分享 >spark的学习-03

spark的学习-03

时间:2024-11-07 20:20:02浏览次数:6  
标签:03 False rdd 分区 学习 RDD print spark True

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())  #9

rdd1.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连接将结果写入MySQL
        print(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)
    # 打印结果:构建RDD
  rs_rdd.foreach(lambda x: print(x))
    # 打印第一行:重新构建一遍
  print(rs_rdd.first())
    # 统计行数:重新构建一遍
    print(rs_rdd.count())

    # todo:3-关闭SparkContext
    time.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的血脉关系的

标签:03,False,rdd,分区,学习,RDD,print,spark,True
From: https://blog.csdn.net/qq_62984376/article/details/143605529

相关文章

  • 大数据学习笔记 第5天 ZooKeeper 3.6.3安装部署 CAP原则 Paxos算法 ZAB协议详解
    ZooKeeper3.6.3重点CAP原则Paxos算法存储模型和监听机制一、集群与分布式集群:将一个任务部署在多个服务器,每个服务器都能独立完成该任务。分布式:将一个任务拆分成若干个子任务,由若干个服务器分别完成这些子任务,每个服务器只能完成某个特定的子任务。从概念上就可......
  • SparkSQL
    一、概述1.1、什么是sparksql  SparkSQL是Spark中用于处理结构化数据的模块,它提供了两个主要的编程抽象:DataFrame和DataSet,并且还可以作为分布式SQL查询引擎使用。SparkSQL的设计目的是简化结构化数据的处理和查询,同时提高执行效率。  传统的HiveSQL通过......
  • 深度学习(三)1.梯度下降
    一、梯度下降 1.梯度下降是什么 梯度下降(GradientDescent)是深度学习中一种至关重要的优化算法,其核心目的是寻找最佳模型参数或权重,从而最小化损失函数。2.基本原理 梯度下降算法的基本思想是:通过迭代的方式,不断调整参数值,沿着目标函数(通常是损失函数)负梯度方向(即函数值......
  • 使用c语言,用keil5进行stm32F103c8点灯实验
    提示:文章写完后,目录可以自动生成,如何生成可参考右边的帮助文档文章目录前言学习stm32首先要学会最基础的电灯实验。进行电灯实验需要进行一些前提工作,需要建立启动文件start和标准外设驱动文件library文件和uesr文件这三个工程文件。具体文件可在网站上进行搜素拷贝......
  • Code Block学习
      CodeBlock 可以用CodeBlock直接输入数字,字符和公式 调用内置函数我们可以在CodeBlock中直接调用内置函数。比如,在空间中创建一个点,我们可以使用Point.ByCoordinates这样的节点,该节点可以通过输入x,y两个值来确定要创建点的位置。Point.ByCoordinates(10,10)......
  • 架构学习中对微服务架构的学习
    微服务架构目录微服务架构一、定义与特点二、优势三、挑战四、应用场景五、实现方式微服务架构(MicroserviceArchitecture)是一种将大型应用程序拆分成一组小的、独立的服务单元的架构模式。每个服务单元运行在独立的进程中,服务之间通过轻量级的通信机制(如HTTPRESTfulAPI)进行......
  • C语言学习8
    最近学的不是很多,只能更几个C语言小游戏给大家看看。一.猜数字游戏(难)1.随机数的生成要想完成猜数字游戏,⾸先得产⽣随机数,那怎么产⽣随机数呢?2.randC语⾔提供了⼀个函数叫rand,这函数是可以⽣成随机数的,函数原型如下所⽰:intrand(void);rand函数会返回⼀个伪随机数......
  • C# SqlSugar学习笔记
    前言今天介绍一款C#ORM框架什么是ORM?来看一下百度术语:对象关系映射(英语:ObjectRelationalMapping,简称ORM,或O/RM,或O/Rmapping),是一种程序技术,用于实现面向对象编程语言里不同类型系统的数据之间的转换 通俗理解ORM我们只需要知道ORM是一种技术,使用了ORM之后我们就不必在......
  • React学习笔记(六) Create React App
    React学习笔记(六)CreateReactApp 1、介绍CreateReactApp是官方支持的创建单页应用程序的方法,可以帮助我们快速搭建一个繁杂的脚手架我们可以直接使用命令 npminstallcreate-react-app-g 全局安装CreateReactApp然后使用命令 create-react-app<project-na......
  • 高性能的Reactor和Proactor模式学习
    0、引言在上一篇的笔记中,我们学习了操作系统提供的高效I/O管理技术,主要用于解决服务器在高并发场景下的资源浪费和瓶颈问题。但是在实际的代码编写中,要是我们都全部调用底层的I/O多路复用接口来编写网络程序这种面向过程的方式,必然会导致开发的效率不高。于是在这一章节,我们来学......