PySpark(三)RDD持久化、共享变量、Spark内核制度,Spark Shuffle、Spark执行流程

目录

RDD持久化

RDD 的数据是过程数据

 RDD 缓存

RDD CheckPoint

共享变量

广播变量

累加器

Spark 内核调度

DAG

DAG 的宽窄依赖和阶段划分 

内存迭代计算 

Spark是怎么做内存计算的? DAG的作用?Stage阶段划分的作用?

Spark为什么比MapReduce快?

Spark并行度 

Spark Shuffle

Hash Shuffle

Sort Shuffle

Spark执行流程


RDD持久化

RDD 的数据是过程数据

RDD之间进行相互迭代计算(Transformation的转换),当执行开启后,新RDD的生成,代表老RDD的消失

RDD的数据是过程数据,只在处理的过程中存在,一旦处理完成,就不见了
这个特性可以最大化的利用资源,老旧RDD没用了 就从内存中清理,给后续的计算腾出内存空间.

例如下面这个例子,生成rdd4的时候, rdd3已经被销毁了,然后下面rdd5需要调用rdd3的时候,只能从rdd->rdd2->rdd3再重新生成一次rdd3

    rdd = sc.parallelize([('a',1),('a',3),('a',6),('b',1),('b',2),('c',1)],3)rdd2 = rdd.map(lambda x:(x[0],x[1]+2))rdd3 = rdd2.distinct()rdd4 = rdd3.filter(lambda x:x[1]>5)print(rdd4.collect())# [('a', 8)]rdd5 = rdd3.glom()print(rdd5.collect())# [[('a', 5), ('a', 8)], [('a', 3), ('b', 4)], [('b', 3), ('c', 3)]]

 RDD 缓存

RDD的缓存技术:Spark提供了缓存AP1,可以让我们通过调用AP1,将指定的RDD数据保留在内存或者硬盘上缓存的API

最开始要引入:from pyspark.storagelevel import StorageLevel

缓存技术可以将过程RDD数据,持久化保存到内存或者硬盘上
但是,这个保存在设定上是认为不安全的,缓存的数据在设计上是 认为 有丢失风险的
所以,缓存有一个特点就是: 其保留RDD之间的血缘(依赖)关系
一旦缓存丢失可以基于血缘关系的记录重新计算这个RDD的数据

缓存如何丢失:
在内存中的缓存是不安全的,比如断电计算任务内存不足把缓存清理给计算让路

硬盘中因为硬盘损坏也是可能丢失的 

RDD缓存采用的是分散存储,也就是每一个executor都会将其处理的部分RDD存放在自己的内存或硬盘中 

RDD CheckPoint

CheckPoint技术也是将RDD的数据保存起来,但是它仅支持硬盘存储
并且:它被设计认为是安全的,不保留 血缘关系

checkPoint存储RDD数据,是集中收集各个分区数据进行存储而缓存是分散存储,也就是说先将executor中的数据收集起来(比如收集到hdfs),然后再进行存储

缓存和CheckPoint的对比

  • CheckPoint不管分区数量多少,风险是一样的,缓存分区越多,风险越高
  • CheckPoint支持写入HDFS,缓存不行HDFS是高可靠存储,checkPoint被认为是安全的.
  • checkPoint不支持内存缓存可以缓存如果写内存性能比checkPoint要好一些
  • CheckPoint因为设计认为是安全的,所以不保留血缘关系,而缓存因为设计上认为不安全,所以保留
    sc.setCheckpointDir("hdfs://node1:8020/checkpoint")#设置存储位置rdd = sc.parallelize([('a',1),('a',3),('a',6),('b',1),('b',2),('c',1)],3)rdd2 = rdd.map(lambda x:(x[0],x[1]+2))rdd3 = rdd2.distinct()rdd3.checkpoint()

共享变量

广播变量

假如有下面一个场景,需要根据stu_ifo将score_ifo中的数字替换为名字,注意这里是本地数据stu_ifo与RDD数据联合处理

    stu_ifo = [(1,'lmx'),(2,'lby'),(3,'lxl')]score_ifo = sc.parallelize([(1,'math',100),(2,'english',87),(1,'english',80),(3,'chinese',98),(2, 'chinese', 68),(1, 'chinese', 88)])def func(data):for i in stu_ifo:if data[0] == i[0]:return (i[1],data[1],data[2])get = score_ifo.map(func)print(get.collect())

一般情况下, 如果一个executor里面有多个分区的情况,那么每个分区都要向driver申请一份本地数据,然而由于executor内部的数据是共享的,这样就会多申请了一份stu_ifo

只需要将stu_ifo标记为广播变量,就可以解决这个问题。只会向一个executor发送一次数据

只需要在之前声明:

stu_if_breadcast = sc.broadcast(stu_ifo)

然后使用的时候:

stu_if_breadcast.value

使用这个方法,可以节省IO次数以及executor内存

 上述代码变为:

    stu_ifo = [(1,'lmx'),(2,'lby'),(3,'lxl')]stu_if_breadcast = sc.broadcast(stu_ifo)score_ifo = sc.parallelize([(1,'math',100),(2,'english',87),(1,'english',80),(3,'chinese',98),(2, 'chinese', 68),(1, 'chinese', 88)])def func(data):for i in stu_if_breadcast.value:if data[0] == i[0]:return (i[1],data[1],data[2])get = score_ifo.map(func)print(get.collect())

累加器

针对下面这种场景,我希望每map一次,我的num加1:

    rdd = sc.parallelize([1,2,3,4,5,6,7,8,9,10],2)def countt(data):global numnum+=1print(num)return dataprint(rdd.map(countt).collect())print(num)# 1# 2# 3# 4# 5# 1# 2# 3# 4# 5# [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]# 0

观察结果,由于分区的情况,在每个executor内num都加到了5,但是最后的num却还是0

因为加1操作只会发生在 executor 中,而最后打印的是driver中的num,所以还是0

这里可以使用spark的累加器变量:

    num = sc.accumulator(0) #累加器rdd = sc.parallelize([1,2,3,4,5,6,7,8,9,10],2)def countt(data):global numnum+=1return datardd.map(countt).collect()print(num)# 10

Spark 内核调度

DAG

DAG:有向无环图

在spark中,每一个 action算子都会将前面的一串rdd依赖链条执行起来,这些执行链条其实就是DAG

有多少个action算子,就有多少个执行链条(JOB),就有多少个DAG

        如果一个代码中,写了n个Action,那么这个代码运行起来产生n个JOB,每个JOB有自己的DAG个代码运行起来,在Spark中称之为: Application

spark中数据都是分区的,所以实际上每一个job都是带有分区关系的DAG

DAG 的宽窄依赖和阶段划分 

RDD的前后关系分为 宽依赖和窄依赖

窄依赖:父RDD的一个分区,全部 将数据发给子RDD的一个分区

宽依赖:父RDD的一个分区将数据发给子RDD的多个分区

宽依赖还有一个别名: shuffle

对于Spark来说会根据DAG,按照宽依赖划分不同的DAG阶段
划分依据:从后向前,遇到宽依赖 就划分出一个阶段称之为stage 

 可以看到,在阶段内都是 窄依赖,这有助于构建内存迭代管道

内存迭代计算 

        在执行上图的程序时,最优的方式肯定是task1,2,3,4,5,6都在独立且单独的线程中完成(还存在另外一种情况,比如b1->p1,p1->下一个p1是不同的线程,那可能会在线程中存在网络IO调用,影响性能)

        task1 中rdd1rdd2 rdd3 的选代计算,都是由一个task(线程完成),这一阶段的这一条线,是纯内存计算.如上图,task1 task2 task3,就形成了三个并行的内存计算管道 

Spark默认受到全局并行度的限制,除了个别算子有特殊分区情况,大部分的算子,都会遵循全局并行度的要求,来规划自己的分区数如果全局并行度是3,其实大部分算子分区都是3
注意:Spark一般推荐只设置全局并行度,不要再算子上设置并行度除了一些排序算子外,计算算子就让他默认开分区就可以了.

Spark是怎么做内存计算的? DAG的作用?Stage阶段划分的作用?

  • Spark会产生DAG图
  • DAG图会基于分区和宽窄依赖关系划分阶段
  • 一个阶段的内部都是 窄依赖,窄依赖内,如果形成前后1:1的分区对应关系就可以产生许多内存选代计算的管道
  • 这些内存迭代计算的管道,就是一个个具体的执行Task
  • 一个Task是一个具体的线程,任务跑在一个线程内,就是走内存计算了 

Spark为什么比MapReduce快?

  • Spark的算子丰富,MapReduce算子匮乏(Map和Reduce),MapReduce这个编程模型,很难在一套MR中处理复杂的任务.很多的复杂任务,是需要写多个MapReduce进行串联多个MR串联通过磁盘交互数据
  • Spark可以执行内存迭代,算子之间形成DAG 基于依赖划分阶段后,在阶段内形成内存选代管道.但是MapReduce的Map和Reduce之间的交互依旧是通过硬盘来交互的

Spark并行度 

Spark的并行: 在同一时间内, 有多少个task在同时运行
并行度:并行能力的设置
比如设置并行度6,其实就是要6个task并行在跑在有了6个task并行的前提下,rdd的分区就被规划成6个分区了

优先级从高到低:
代码中
客户端提交参数中
配置文件中
默认(1,但是不会全部以1来跑,多数时候基于读取文件的分片数量来作为默认并行度)

 集群如何规划并行度?

结论:设置为CPU总核心的2~10倍

        为什么要设置最少2倍?
        CPU的一个核心同一时间只能干一件事情所以,在100个核心的情况下,设置100个并行,就能让CPU 100%出力

        这种设置下如果task的压力不均衡,某个task先执行完了就导致某个CPU核心空闲

        所以,我们将Task(并行)分配的数量变多,比如800个并行,同一时间只有100个在运行,700个在等待.但是可以确保某个task运行完了.后续有task补上,不让cpu闲下来,最大程度利用集群的资源 

Spark Shuffle

Spark在DAG调度阶段会将一个Job划分为多个Stage,上游Stage做map工作,下游Stage做reduce工作,其本质上 还是MapReduce计算框架。Shuffle是连接map和reduce之间的桥梁,它将map的输出对应到reduce输入中,涉及 到序列化反序列化、跨节点网络IO以及磁盘读写IO等。

 Spark的Shuffle分为Write和Read两个阶段,分属于两个不同的Stage,前者是Parent Stage的最后一步,后者是 Child Stage的第一步。

在Spark的中,负责shuffle过程的执行、计算和处理的组件主要就是ShuffleManager,也即shuffle管理器。ShuffleManager随着 Spark的发展有两种实现的方式,分别为HashShuffleManager和SortShuffleManager,因此spark的Shuffle有Hash ShuffleSort Shuffle两种。 

Hash Shuffle

未经优化的hashShuffleManager: 

        HashShuffle是根据task的计算结果的key值的hashcode%ReduceTask来决定放入哪一个区分,这样保证相同的数据 一定放入一个分区,根据下游的task决定生成几个文件,先生成缓冲区文件在写入磁盘文件,再将block文件进行合并。

        一个task内部会根据hash分类,然后将不同类的数据放入不同的磁盘文件,未经优化的shuffle write操作所产生的磁盘文件的数量是极其惊人的,也就是下图所产生的的block file

 优化的hashShuffleManager:

        在shuffle write过程中,task就不是为下游stage的每个task创建一个磁盘文件了。此时会出现shuffleFileGroup的概 念,每个shuffleFileGroup会对应一批磁盘文件,每一个Group磁盘文件的数量与下游stage的task数量是相同的。

        其实说到底就是其在executor内部就会进行数据的汇合操作,大大减少了磁盘文件的生成

Sort Shuffle

(1)该模式下,数据会先写入一个内存数据结构中(默认5M),此时根据不同的shuffle算子,可能选用不同的数据结构。如果是reduceByKey这种聚合类的shuffle算子,那么会选用Map数据结构,一边通过Map进行聚合,一边写入内 存;如果是join这种普通的shuffle算子,那么会选用Array数据结构,直接写入内存。

(2)接着,每写一条数据进入内存数据结构之后,就会判断一下,是否达到了某个临界阈值。如果达到临界阈值的话 ,那么就会尝试将内存数据结构中的数据溢写到磁盘,然后清空内存数据结构。

(3)排序 在溢写到磁盘文件之前,会先根据key对内存数据结构中已有的数据进行排序。 (4)溢写 排序过后,会分批将数据写入磁盘文件。默认的batch数量是10000条,也就是说,排序好的数据,会以每批1万条数 据的形式分批写入磁盘文件。

(5)merge 一个task将所有数据写入内存数据结构的过程中,会发生多次磁盘溢写操作,也就会产生多个临时文件。最后会将之 前所有的临时磁盘文件都进行合并成1个磁盘文件,这就是merge过程。 由于一个task就只对应一个磁盘文件,也就意味着该task为Reduce端的stage的task准备的数据都在这一个文件中, 因此还会单独写一份索引文件,其中标识了下游各个task的数据在文件中的start offset与end offset。


说到底最大的区别就是,一个task只会生成一个磁盘文件和一个索引文件,大大降低了磁盘占有和网络IO数量

Sort Shuffle bypass机制

bypass运行机制的触发条件如下:

1)shuffle map task数量小于spark.shuffle.sort.bypassMergeThreshold=200参数的值。

2)不是map combine聚合的shuffle算子(比如reduceByKey有map combie)(这种算子不需要排序)

        此时task会为每个reduce端的task都创建一个临时磁盘文件,并将数据按key进行hash,然后根据key的hash值, 将key写入对应的磁盘文件之中。当然,写入磁盘文件时也是先写入内存缓冲,缓冲写满之后再溢写到磁盘文件的 。最后,同样会将所有临时磁盘文件都合并成一个磁盘文件,并创建一个单独的索引文件。

        该过程的磁盘写机制其实跟未经优化的HashShuffleManager是一模一样的,因为都要创建数量惊人的磁盘文件, 只是在最后会做一个磁盘文件的合并而已。因此少量的最终磁盘文件,也让该机制相对未经优化的 HashShuffleManager来说,shuffle read的性能会更好


而该机制与普通SortShuffleManager运行机制的不同在于:

第一,磁盘写机制不同;

第二,不会进行排序。也就是说,启用该机制的最大好处在于,shuffle write过程中,不需要进行数据的排序操作, 也就节省掉了这部分的性能开销。

Spark执行流程

在大方向上:

  1. 提交代码
  2. 生成Driver ,DAG Scheduler规划逻辑任务
  3. 生成Executor(被Driver生成)
  4. Driver内TaskScheduler去监控整个Spark程序的执行

在细节上,以YARN为例:

启动ApplicationMaster
AM启动Driver
- Driver构建DAG调度器规划任务
- Driver和AM通讯.AM得知要多少容器去申请
- Driver在申请的容器内部启动Executor
- Driver内的Task调度器,调度任务执行 

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/675648.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

Could not connect to Redis at 127.0.0.1:6379:由于目标计算机积极拒绝,无法连接...问题解决方法之一

一、问题描述 将Redis压缩包解压后,安装Redis过程中出现问题Could not connect to Redis at 127.0.0.1:6379:由于目标计算机积极拒绝,无法连接... 官网windows下redis开机自启动的指令如下: 1、在redis目录下执行 redis-server --service-in…

GEE Colab——如何利用Matplotlib在colab中进行图形制作

在colab中绘制图表 笔记本的一个常见用途是使用图表进行数据可视化。Colaboratory 提供多种图表工具作为 Python 导入,让这一工作变得简单。 Matplotlib Matplotlib 是最常用的图表工具包,详情请查看其文档,并通过示例获得灵感。 线性图 线性图是一种常见的图表类型,用…

按键扫描16Hz-单片机通用模板

按键扫描16Hz-单片机通用模板 一、按键扫描的原理1、直接检测高低电平类型2、矩阵扫描类型3、ADC检测类型二、key.c的实现1、void keyScan(void) 按键扫描函数①void FHiKey(void) 按键按下功能②void FSameKey(void) 按键长按功能③void FLowKey(void) 按键释放功能三、key.h的…

JavaScript综合练习2

JavaScript 综合练习 2 1. 案例演示 2. 代码实现 <!DOCTYPE html> <html lang"en"><head><meta charset"UTF-8" /><meta name"viewport" content"widthdevice-width, initial-scale1.0" /><title&…

Springboot根据环境读取application配置文件

目录 1. 首先创建两个不同配置文件 2. pom.xml 配置文件 3. 指定环境 4. 最后启动测试 1. 首先创建两个不同配置文件 分别为开发环境和生产环境 application-dev.properties 和 application-prod.properties application-dev.properties 配置为 1931 端口 application-pro…

放假--寒假自学版 day1(补2.5)

fread 函数&#xff1a; 今日练习 C语言面试题5道~ 1. static 有什么用途&#xff1f;&#xff08;请至少说明两种&#xff09; 1) 限制变量的作用域 2) 设置变量的存储域 2. 引用与指针有什么区别&#xff1f; 1) 引用必须被初始化&#xff0c;指针不必。 2) 引用初始…

【大数据】Flink 中的 Slot、Task、Subtask、并行度

Flink 中的 Slot、Task、Subtask、并行度 1.并行度2.Task 与线程3.算子链与 slot 共享资源组4.Task slots 与系统资源5.总结 我们在使用 Flink 时&#xff0c;经常会听到 task&#xff0c;slot&#xff0c;线程 以及 并行度 这几个概念&#xff0c;对于初学者来说&#xff0c;这…

【网工】华为设备命令学习(服务器发布)

本次实验主要是内网静态nat配置没&#xff0c;对外地址可以理解为一台内网的服务器&#xff0c;外网设备可以ping通内网的服务器设备&#xff0c;但是ping不通内网的IP。 除了AR1设备配置有区别&#xff0c;其他设备都是基础IP的配置。 [Huawei]int g0/0/0 [Huawei-GigabitEt…

双指针和单调栈

双指针 用于解决一类基于子段的统计问题 子段就是&#xff1a;数组中连续的一段 可以用一个闭区间来表示数组中的连续一段 这个方法核心就是优化&#xff1a;两种循环的枚举 也就是枚举左端点l和右端点r的所有可能优化关键就是&#xff1a;去除枚举中的冗余部分 具体优化策略…

设计模式-行为型模式(下)

1.访问者模式 访问者模式在实际开发中使用的非常少,因为它比较难以实现并且应用该模式肯能会导致代码的可读性变差,可维护性变差,在没有特别必要的情况下,不建议使用访问者模式. 访问者模式(Visitor Pattern) 的原始定义是&#xff1a; 允许在运行时将一个或多个操作应用于一…

【linux开发工具】vim详解

&#x1f4d9; 作者简介 &#xff1a;RO-BERRY &#x1f4d7; 学习方向&#xff1a;致力于C、C、数据结构、TCP/IP、数据库等等一系列知识 &#x1f4d2; 日后方向 : 偏向于CPP开发以及大数据方向&#xff0c;欢迎各位关注&#xff0c;谢谢各位的支持 “学如逆水行舟&#xff0…

自适应二次元404页面源码

自适应二次元404页面源码&#xff0c;HTMLCSSJS,喜欢二次元的朋友可以下载使用 蓝奏云&#xff1a;https://wfr.lanzout.com/iuPNQ1ns7dxg

多模态对比语言图像预训练CLIP:打破语言与视觉的界限,具备零样本能力

多模态对比语言图像预训练CLIP:打破语言与视觉的界限,具备零样本能力。 一种基于多模态(图像、文本)对比训练的神经网络。它可以在给定图像的情况下,使用自然语言来预测最相关的文本片段,而无需为特定任务进行优化。CLIP的设计类似于GPT-2和GPT-3,具备出色的零射击能力…

API接口访问鉴权设计和实现的经验总结

API接口访问鉴权是保护API资源安全的重要措施。本文总结了一些常见的API接口访问鉴权设计和实现方法&#xff0c;以帮助开发人员更好地理解和应用这些技术。 1. 什么是API接口访问鉴权&#xff1f; - 解释了API接口访问鉴权的基本概念和作用&#xff0c;以及为什么需要对A…

R语言入门笔记2.0

1.创建数据框 在R语言中&#xff0c;可以使用data.frame函数来创建数据框。以下是一个简单的示例&#xff0c;这段R语言代码创建了一个名为student的数据框&#xff0c;其中包含了学生的ID、性别、姓名和出生日期&#xff0c;首先创建一个包含学生出生日期的向量&#xff0c;再…

基于华为云欧拉操作系统(HCE OS)容器化部署传统应用(Redis+Postgresql+Git+SpringBoot+Nginx)

写在前面 博文内容为 华为云欧拉操作系统入门级开发者认证(HCCDA – Huawei Cloud EulerOS)实验笔记整理认证地址&#xff1a;https://edu.huaweicloud.com/certificationindex/developer/9bf91efb086a448ab4331a2f53a4d3a1博文内容涉及一个传统 Springboot 应用HCE部署&#x…

文件上传-Webshell

Webshell简介 webshell就是以aspphpjsp或者cgi等网页文件形式存在的一种命令执行环境&#xff0c;也可以将其称做为一种网页木马后门。 攻击者可通过这种网页后门获得网站服务器操作权限&#xff0c;控制网站服务器以进行上传下载文件、查看数据库、执行命令等… 什么是木马 …

【Docker】02 镜像管理

文章目录 一、Images镜像二、管理操作2.1 搜索镜像2.1.1 命令行搜索2.1.2 页面搜索2.1.3 搜索条件 2.2 下载镜像2.3 查看本地镜像2.3.1 docker images2.3.2 --help2.3.3 repository name2.3.4 --filter2.3.5 -q2.3.6 --format 2.4 给镜像打标签2.5 推送镜像2.6 删除镜像2.7 导出…

IP代理在网络中解决了哪些问题?代理IP使用时效是什么意思?

随着互联网的普及和发展&#xff0c;IP代理作为一种网络工具&#xff0c;被广泛应用于各种场景。IP代理的使用可以解决很多网络中的问题&#xff0c;提高网络访问的速度和安全性。本文将详细介绍IP代理在网络中解决的问题&#xff0c;以及代理IP使用时效的含义。 一、IP代理在网…

Unity入门学习

目录 Unity环境搭建Unity引擎是什么软件下载和安装工程文件夹 Unity界面基础Scene场景和Hierarchy层级窗口Game游戏和Project工程Inspector和Console工具栏和父子关系 Unity工作原理反射机制和游戏场景预设体和资源包的导入导出 Unity脚本基础脚本基本规则生命周期函数Inspecto…