Spark RDD优化

Spark RDD优化

  • 一、分区优化
  • 二、持久化优化
  • 三、依赖优化
  • 四、共享变量优化
  • 五、提交模式与运行模式优化
  • 六、其他优化

一、分区优化

  • 分区数调整:RDD的分区数可以通过repartitioncoalesce方法进行调整。合理的分区数可以提高并行度,但过多的分区会增加管理开销。通常,分区数应根据数据规模和集群资源进行调整。

    val rdd: RDD[String] = rdd.coalesce(numPartitions:Int, shuffle:Boolean)
    val rdd: RDD[String] = rdd.repartition(numPartitions:Int) 
    // repartition(numPartitions: Int) 等价于 coalesce(numPartitions, true) 
    
    1. 缩小分区

      存在过多的小任务的时候收缩合并分区,减少分区的个数,减少任务调度成本
      默认情况下,不会对数据重组,比如:3个合成2个,采用 {1+2},{3},容易导致数据倾斜
      若需数据均衡,则将 shuffle 参数设置为 true 即可

    2. 扩大分区

      若需要扩大分区,shuffle 参数必须设置为 true
      若将2个分区拆分成3个,必须打乱重新分区,否则数据还是在两个分区(有一个分区为空),{1},{2},{空}

  • 数据本地性:Spark会尽量将数据分配给与数据源相同的计算节点上,以减少数据移动的开销。在创建RDD时,可以通过设置分区偏好(如preferredLocations)或自定义分区来优化数据本地性,以最小化网络传输并最大化计算效率。

    自定义分区

    // 自定义分区器
    class MyPartitioner(numPartitions: Int) extends Partitioner {override def numPartitions: Int = numPartitions   // 返回分区器的分区数量override def getPartition(key: Any): Int = {// 这里需要实现分区逻辑// 返回值是一个整数,表示该键应该被分配到哪个分区}
    }
    
    // 使用自定义分区器重新分区  
    val partitionedRDD = rdd.partitionBy(new MyPartitioner(2))  // 传入分区个数
    
  • 处理数据倾斜:数据倾斜是指某些分区包含的数据远远多于其他分区,导致计算资源分配不均。可以使用repartitioncoalesce方法重新分区RDD,或使用reduceByKeygroupByKey的变体等特定操作来减轻数据倾斜的影响。

二、持久化优化

  • 持久化策略:对于需要多次使用的RDD,应该进行持久化操作,以避免重复计算。持久化策略包括内存持久化(如MEMORY_ONLY)、磁盘持久化(如DISK_ONLY)以及内存和磁盘混合持久化(如MEMORY_AND_DISK)等。

  • 序列化:使用序列化可以进一步减少内存消耗,并提高持久化效率。Spark支持多种序列化框架,如Java序列化、Kryo序列化等。Kryo序列化通常比Java序列化更快,且占用空间更小。

    // 临时存储于【xx】重用,job结束后自动删除 
    val rddCache: RDD[T] = rdd.cache()					// 到内存上
    val rdd: RDD[T] = rdd.persist(level:StorageLevel)
    // cache() 		等价于persist(StorageLevel.MEMORY_ONLY)
    // persisit() 	参数如下
    
    StorageLevel.MEMORY_ONLY				只写到内存上
    StorageLevel.DISK_ONLY					只写到磁盘上
    StorageLevel.OFF_HEAP					使用堆外内存
    StorageLevel.MEMORY_AND_DISK			先内存,后磁盘 
    StorageLevel.MEMORY_AND_DISK_SER		先内存,后磁盘,采取序列化方式
    StorageLevel.MEMORY_AND_DISK_SER_2 		先内存,后磁盘,采取二代序列化方式
    
  • 检查点:对于需要长时间运行或可能遭受故障的应用,设置检查点(Checkpoint)可以将RDD的状态保存到稳定存储中,以便在故障后恢复。检查点会切断RDD的血统关系,从而避免重新计算整个血统链。

    // checkpoint 长久存储于【磁盘】重用,job结束后不会删除,涉及IO性能较差,安全且一般和cache组合使用
    val conf = new SparkConf().setAppName("spark_rdd").setMaster("local[4]")
    val sc = SparkContext.getOrCreate(conf)
    // 设置检查点路径
    sc.setCheckpointDir("hdfs://ip:9000/spark/checkpoint")
    // ... 
    rdd.checkpoint()	// 将该 RDD 的内容写入到设置的路径,并在该 RDD 的计算图中插入一个检查点(Checkpoint)节点
    

三、依赖优化

  • 宽依赖与窄依赖:RDD之间的依赖关系分为宽依赖和窄依赖。窄依赖有助于实现数据本地性,而宽依赖则可能导致数据移动和网络开销。在设计RDD转换操作时,应尽量避免不必要的宽依赖。

    1、Driver程序提交后

    1、Spark调度器将所有的RDD看成是一个Stage
    2、然后对此Stage进行逆向回溯,遇到Shuffle就断开,形成一个新的Stage
    3、遇到窄依赖,则归并到同一个Stage
    4、等到所有的步骤回溯完成,便生成一个DAG图

    2、为什么要划分阶段

    1、基于数据的分区,本着传递计算的性能远高于传递数据,所以数据本地化是提升性能的重要途径之一
    2、一组串行的算子,无需 Shuffle,基于数据本地化可以作为一个独立的阶段连续并行执行
    3、经过一组串行算子计算,遇到 Shuffle 操作,默认情况下 Shuffle 不会改变分区数量,但会因为 numPartitions:Int, partitioner:Partitioner 等参数重新分配,过程数据会【写盘供子RDD拉取(类MapReduce)】

    3、RDD依赖关系

    • Lineage:血统、遗传

      RDD最重要的特性之一,保存了RDD的依赖关系

      RDD实现了基于Lineage的容错机制

    • 依赖关系 org.apache.spark.Dependency

      窄依赖 NarrowDependency,1V1 OneToOneDependency,1VN RangeDependency
      宽依赖 ShuffleDependency

    • 当RDD分区丢失时

      对于窄依赖,Spark只需要重新计算丢失分区的父RDD分区即可。
      对于宽依赖,Spark需要重新执行整个shuffle过程,以重新生成丢失的数据。
      若配合持久化更佳:cache, persist, checkpoint

    在这里插入图片描述

    类型
    窄依赖map,flatMap,mapPartitions,mapPartitionsWithIndex,glom,filter,distinct,intersection,sample,union,subtract,zip…,cogroup
    宽依赖sortBy,sortByKey,groupByKey,reduceByKey,cogroup,join,partitionBy,repartition
    不一定的情况在Spark中,并非所有操作都可以明确地归类为宽依赖或窄依赖。有些操作可能根据具体的实现或上下文而有所不同。然而,在大多数情况下,上述提到的算子可以清晰地划分为宽依赖或窄依赖。
    如:reduceByKey(【partitioner: Partitioner】, func: (V, V) => V)
    若使用的是带 partitioner 的重载且 Partitioner 和父RDD的 Partitioner一致
    则为窄依赖RDD,否则为宽依赖ShuffledRDD
    
  • 优化转换操作:在可能的情况下,使用能够减少shuffle操作的转换函数,如mapPartitions代替mapreduceByKey代替groupByKey等。这些操作可以减少数据在网络中的传输量,从而提高性能。

    shuffle性能较差:因为shuffle必须落盘,内存中等数据会OOM
    groupByKey只分组(存在Shuffle) + reduce只聚合<=结果同,性能不同=>
    reduceByKey先分组、预聚合、再聚合(存在Shuffle) 
    

四、共享变量优化

  • 广播大变量:当Spark作业中需要使用到较大的外部变量时,可以将这些变量广播到每个节点的Executor上,而不是每个Task都复制一份。这样可以减少网络传输开销和内存消耗。

    val bc:Broadcast[T] = sc.broadcast(value:T)		// 创建广播变量  
    rdd.mapPartitions(itPar=>{val v:T = bc.value	// 在每个分区内部,通过bc.value获取广播变量的值  ...					// 使用v进行计算...
    })
    
  • 累加器(Accumulators):累加器提供了一种有效的手段来进行分布式计算中的统计和计数操作,减少通信开销,并简化聚合操作。

    累加器:accumulate:只能 add 操作,常用于计数
    1、定义在Driver端的一个变量,Cluster中每一个Task都会有一份Copy
    2、所有的Task都计算完成后,将所有Task中的Copy合并到驱动程序中的变量中
    非累加器:在所有Task中的都会是独立Copy,不会有合并

    累加器
    val accLong: LongAccumulator = sc.longAccumulator("longAcc")	// 定义累加器
    val accDouble: DoubleAccumulator = sc.doubleAccumulator("doubleAcc")
    rdd.mapPartitions(itPar=>{...accLong.add(v:Long)		// 将值添加到累加器中accDouble.add(v:Double)...
    })
    accXxx.reset()		// 重置累加器
    val isZero:Boolean = accXxx.isZero	// 检查累加器是否为零值
    val num:Long|Double = accXxx.value|sum|count|avg // 获取累加器的值、总和、计数或平均值
    
    // 定义一个累加器,用于统计 "bad" 记录的数量
    val errorCount = sc.longAccumulator("Error Count")
    val data = sc.parallelize(Array("good", "bad", "good", "bad", "good"))
    data.foreach(record => if (record == "bad") errorCount.add(1))
    // 打印累加器的值,即 "bad" 记录的总数println(s"Total errors: ${errorCount.value}")
    

    自定义累加器:

    写一个类继承 import org.apache.spark.util.AccumulatorV2[IN, OUT]

    abstract class AccumulatorV2[IN, OUT] extends Serializable {// 返回是否为零值累加器def isZero: Boolean// 创建此累加器的新副本,其为零值def copyAndReset(): AccumulatorV2[IN, OUT] = {...}// 创建此累加器的新副本def copy(): AccumulatorV2[IN, OUT]// 重置此累加器为零值def reset(): Unit// 添加:接收输入并累加def add(v: IN): Unit// 合并:合并另一个相同类型的累加器并更新其状态def merge(other: AccumulatorV2[IN, OUT]): Unit// 当前累加器的值def value: OUT
    }
    
  • 自定义计量器优化(Custom Metrics):自定义计量器允许用户定义和收集特定的性能指标,提供更细粒度的作业监控和调优能力。通过 SparkListener 接口,可以实现自定义的监听器来监控和记录所需的指标。

五、提交模式与运行模式优化

  • 提交模式:Spark支持Client模式和Cluster模式两种提交方式。Client模式便于查看日志和结果,但可能消耗较多资源;Cluster模式则更适合大规模作业,但查看日志和结果可能不太方便。应根据实际情况选择合适的提交模式。

    spark-submit --class <MainClass> --master <MasterURL> --deploy-mode <DeployMode> <PathToJar>
    

    <MainClass>:包含 main 方法的主类的名称。

    <MasterURL>:指定集群的 Master URL。

    <DeployMode>:指定提交模式,可以是 clientcluster

    <PathToJar>:包含 Spark 应用程序的 JAR 文件的路径。

    spark-submit --class SparkClientModeApp --master yarn --deploy-mode client /path/to/your/jarfile.jar	
    spark-submit --class SparkClientModeApp --master yarn --deploy-mode cluster /path/to/your/jarfile.jar
    
  • 运行模式:Spark支持多种运行模式,如Local模式、Standalone模式、YARN模式等。不同的运行模式适用于不同的场景和需求。例如,Local模式适用于本地开发和测试;Standalone模式适用于构建独立的Spark集群;YARN模式则适用于与Hadoop生态系统集成。

    local: 在单核上运行
    local[N]: 在指定数量的 N 个核上运行,如 “local[4]”
    local[*]: 使用所有可用的核
    spark://HOST:PORT: 连接到指定的 Spark standalone cluster
    yarn: 连接到 YARN 集群
    mesos://HOST:PORT: 连接到 Mesos 集群

六、其他优化

  • 序列化框架选择:除了Kryo序列化外,还可以考虑使用其他高效的序列化框架来优化Spark作业的性能。
  • 监控与调优:使用Spark提供的监控工具和API(如Spark UI、getStorageLevel方法等)来监控作业的运行状态和性能瓶颈,并根据监控结果进行调优。

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

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

相关文章

捷配PCB 6个PCB板材关键参数解读技巧

PCB板材是指覆铜基板&#xff0c;是制造电路板的最主要材料。 板材的一些关键性能参数对电路板的生产加工、元器件贴装焊接、电子产品的功能实现以及产品的使用环境或寿命等都将产生一定程度的影响&#xff0c;所以掌握板材的关键参数在实际应用中非常有必要。 PCB板材的关键性…

Nifi内置处理器Processor的开发

Nifi-Processor自定义开发的流程 之前说过&#xff0c;大部分的数据处理&#xff0c;我们可以基于ExcuseGroovyScript处理器&#xff0c;编写Groovy脚本去完成&#xff08;或者Jpython&#xff0c;Js脚本等对应的组件&#xff09;&#xff0c;只能说这是基于Nifi平台的使用层面…

UML建模案例分析-时序图和类图的消息传递

概念 类图和时序图之间的交互是通过消息&#xff0c;即成员函数的调用体现的。但要遵循一定的原则&#xff0c;可参见&#xff1a; 面向对象原则之一&#xff0c;面相对象原则之二。 例子 一个电子商务系统&#xff0c;会员可通过电子商务系统购买零件。具体功能需求如下&am…

【中项第三版】系统集成项目管理工程师 | 第 4 章 信息系统架构⑤ | 4.8 - 4.9

前言 第4章对应的内容选择题和案例分析都会进行考查&#xff0c;这一章节属于技术相关的内容&#xff0c;学习要以教材为准。本章分值预计在4-5分。 目录 4.8 云原生架构 4.8.1 发展概述 4.8.2 架构定义 4.8.3 基本原则 4.8.4 常用架构模式 4.8.5 云原生案例 4.9 本章…

如何解决英国Facebook直播网络延时问题?

许多商家在英国进行Facebook直播&#xff0c;但网络延时和卡顿问题常常困扰着用户。这不仅影响观众的观看体验&#xff0c;也会给商家带来巨大损失。本文将探讨解决英国Facebook直播网络延时和卡顿问题的方案&#xff0c;以促进业务发展并提升用户满意度。 海外直播的挑战 海外…

【Flask从入门到精通:第一课:flask的基本介绍、flask快速搭建项目并运行】

从0开始入手到上手一个新的框架&#xff0c;应该怎么展开&#xff1f;flask这种轻量级的框架与django这种的重量级框架的区别&#xff1f;针对web开发过程中&#xff0c;常见的数据库ORM的操作。跟着学习flask的过程中&#xff0c;自己去学习和了解一个新的框架&#xff08;San…

unity中我想实现现实中琴弦的那种互动抖动效果,谈谈思路。

&#x1f3c6;本文收录于《CSDN问答解惑》专栏&#xff0c;主要记录项目实战过程中的Bug之前因后果及提供真实有效的解决方案&#xff0c;希望能够助你一臂之力&#xff0c;帮你早日登顶实现财富自由&#x1f680;&#xff1b;同时&#xff0c;欢迎大家关注&&收藏&…

BouncyCastleProvider 对 X.509 证书的生成

文章目录 前言BouncyCastleProvider 对 X.509 证书的生成1. demo 前言 如果您觉得有用的话&#xff0c;记得给博主点个赞&#xff0c;评论&#xff0c;收藏一键三连啊&#xff0c;写作不易啊^ _ ^。   而且听说点赞的人每天的运气都不会太差&#xff0c;实在白嫖的话&#xf…

程序员学CFA——经济学(六)

经济学&#xff08;六&#xff09; 国际贸易与资本流动国际贸易相关术语开放/封闭经济自由贸易/贸易保护贸易比价国内生产总值与国民生产总值 国际贸易的利弊分析益处弊端 从贸易中获益&#xff1a;比较优势比较优势和绝对优势比较优势的来源 贸易限制和贸易保护施行贸易保护政…

如何快速找车?一文教你如何打造一个高效的停车场反向寻车系统?

在当今快速发展的城市化进程中&#xff0c;大型商业综合体、购物中心和办公楼的停车场规模日益庞大&#xff0c;车辆数量急剧增加。随之而来的是车主在这些大型停车场中寻找自己车辆的困难。由于停车场空间大、结构复杂&#xff0c;缺乏有效的参照物&#xff0c;车主常常在寻车…

基于SpringBoot的网上书城管理系统

你好呀&#xff0c;我是计算机学姐码农小野&#xff01;如果有相关需求&#xff0c;可以私信联系我。 开发语言&#xff1a;Java 数据库&#xff1a;MySQL 技术&#xff1a;Java技术&#xff0c;基于SpringBoot框架 工具&#xff1a;Eclipse&#xff0c;MySQL 系统展示 首…

英福康INFICON FabGuard传感器集成与分析系统PPT

英福康INFICON FabGuard传感器集成与分析系统PPT

VS安装Qt扩展工具

1-Visual Studio中安装QT插件 **插件下载地址&#xff1a;**http://download.qt.io/development_releases/vsaddin/ 关闭VS,双击下载的QT插件&#xff0c;默认安装即可&#xff1b; &#xff08;1&#xff09;配置Qt的MSVC编译器安装路径 打开Visual Studio&#xff0c;在菜单栏…

pbootCMS 数据库sqlite转mysql数据库

前言 pbootCMS默认使用 sqlite数据库 &#xff0c;那么什么是sqlite数据库呢&#xff1f; SQLite&#xff0c;是一款轻型的数据库&#xff0c;是遵守ACID的关系型数据库管理系统&#xff0c;它包含在一个相对小的C库中。它是D.RichardHipp建立的公有领域项目。它的设计目标是嵌…

帕金森患者营养小贴士

&#x1f44b;亲爱的小伙伴们&#xff0c;今天我要给大家带来一份特别的关怀——关于帕金森患者的营养小贴士&#x1f4d8;。帕金森病虽然是一种神经系统疾病&#xff0c;但合理的营养摄入对于患者的生活质量有着不可忽视的影响哦&#xff01;&#x1f4aa; &#x1f34e;多吃水…

万界星空科技日化行业MES解决方案

日化行业MES&#xff08;制造执行系统&#xff09;解决方案是针对日化行业特点而设计的一套全面的生产管理系统&#xff0c;旨在提高生产效率、优化资源配置、加强质量控制&#xff0c;并推动企业的数字化转型。以下是对日化行业MES解决方案的详细阐述&#xff1a; 一、MES解决…

TikTok短视频矩阵系统源码:创新驱动的内容生态引擎

在全球化的数字媒体时代&#xff0c;短视频已成为连接创作者与观众的桥梁。TikTok作为全球领先的短视频社交平台&#xff0c;其背后的短视频矩阵系统源码是支撑这一平台创新与成长的关键。本文将深入探讨TikTok短视频矩阵系统源码如何通过多账号管理、AI自动生成文案、批量剪辑…

【以史为镜、以史明志,知史爱党、知史爱国】中华上下五千年之-唐朝

唐朝&#xff08;618年—907年&#xff09;&#xff0c;是继隋朝之后的大一统中原王朝&#xff0c;共历二十一帝&#xff0c;享国二百八十九年。 隋末天下群雄并起&#xff0c;617年 唐国公李渊于晋阳起兵&#xff0c;次年称帝建立唐朝&#xff0c;定都长安。 唐太宗继位后开创…

经典 SQL 数据库笔试题及答案整理

最近有蛮多小伙伴在跳槽找工作&#xff0c;但对于年限稍短的软件测试工程师&#xff0c;难免会需要进行笔试&#xff0c;而在笔试中&#xff0c;基本都会碰到一道关于数据库的大题&#xff0c;今天这篇文章呢&#xff0c;就收录了下最近学员反馈上来的一些数据库笔试题&#xf…

【高中数学/幂函数】比较a=2^0.3,b=3^0.2,c=7^0.1的大小

【问题】 比较a2^0.3,b3^0.2,c7^0.1的大小 【解答】 a2^0.32^3/10(2^3)^1/108^1/10 b3^0.23^2/10(3^2)^1/109^1/10 c7^0.17^1/10 由于yx^1/10在x正半轴是增函数&#xff0c;底数大的得数就大。 因为9>8>7,所以b>a>c 【图像】 在图像上绘出曲线yx^1/10&…