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.rhkb.cn/news/376960.html

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

相关文章

HarmonyOS 界面开发基础篇

一、布局思路 ArkUI(方舟开发框架)是一套 构建 鸿蒙应用 界面 的框架。 构建页面的最小单位就是“组件”。 组件分类 基础组件&#xff1a;界面呈现的基础元素。 如&#xff1a;文字、图片、按钮等。容器组件&#xff1a;控制布局排布。 如&#xff1a;Row行、Column列等。…

Window10下安装WSL-Ubuntu20.04

1.开启并更新WSL 1.1开启WSL 首先先来看一下电脑是否能够开启WSL:待补充... 然后再来看一下如何开启WSL:win->设置->应用->应用和功能->程序和功能&#xff0c;如下所示&#xff1a; 最后选择启用或关闭Windows功能&#xff0c;开启两个选项&#xff1a;1.Hyper-V…

C++中链表的底层迭代器实现

大家都知道在C的学习中迭代器是必不可少的&#xff0c;今天我们学习的是C中的链表的底层迭代器的实现&#xff0c;首先我们应该先知道链表的底层迭代器和顺序表的底层迭代器在实现上有什么区别&#xff0c;为什么顺序表的底层迭代器更加容易实现&#xff0c;而链表的底层迭代器…

Golang | Leetcode Golang题解之第235题二叉搜索树的最近公共祖先

题目&#xff1a; 题解&#xff1a; func lowestCommonAncestor(root, p, q *TreeNode) (ancestor *TreeNode) {ancestor rootfor {if p.Val < ancestor.Val && q.Val < ancestor.Val {ancestor ancestor.Left} else if p.Val > ancestor.Val && q…

Dify中的工具

Dify中的工具分为内置工具&#xff08;硬编码&#xff09;和第三方工具&#xff08;OpenAPI Swagger/ChatGPT Plugin&#xff09;。工具可被Workflow&#xff08;工作流&#xff09;和Agent使用&#xff0c;当然Workflow也可被发布为工具&#xff0c;这样Workflow&#xff08;工…

解决fidder小黑怪倒出JMeter文件缺失域名、请求头

解决fidder小黑怪倒出JMeter文件缺失域名、请求头 1、目录结构&#xff1a; 2、代码 coding:utf-8 Software:PyCharm Time:2024/7/10 14:02 Author:Dr.zxyimport zipfile import os import xml.etree.ElementTree as ET import re#定义信息头 headers_to_extract [Host, Conn…

芋道框架万字详解(前后端分离)、若依框架、yudao-cloud保姆级攻略

♥️作者&#xff1a;小宋1021 &#x1f935;‍♂️个人主页&#xff1a;小宋1021主页 ♥️坚持分析平时学习到的项目以及学习到的软件开发知识&#xff0c;和大家一起努力呀&#xff01;&#xff01;&#xff01; &#x1f388;&#x1f388;加油&#xff01; 加油&#xff01…

STM32MP135裸机编程:定时器内核时钟频率计算方法

0 工具准备 STM32MP13xx参考手册 1 定时器内核时钟频率计算方法 1.1 定时器分组 STM32MP135的定时器按照时钟源不同分成了三组&#xff0c;如下&#xff1a; APB1: APB2: APB6&#xff1a; 1.2 定时器内核时钟频率计算方法 APB1DIV是APB1的分频系数&#xff0c;APB2DIV、…

51单片机9(使用左移实现流水灯编程)

一、序言&#xff1a;下面我们来给大家介绍一下这个流水灯&#xff0c;流水灯如何来实现&#xff1f;我们依然使用这个工程来完成它。 1、那要使用实现这个流水灯&#xff0c;那我们只需要让D1到D8逐个的点亮&#xff0c;那同样要实现它足够的点亮&#xff0c;也会涉及到延时&…

html设计(两种常见的充电效果)

第一种 完整代码&#xff1a; <!DOCTYPE html> <html lang"en"> <head><meta charset"UTF-8"><meta name"viewport" content"widthdevice-width, initial-scale1.0"><title>Document</title&…

tomcat和nginx实现动静分离

访问nginx就是静态页面&#xff0c;nginx代理index.jsp可以访问tomcat的动态页面。 实验 1、设备以及IP地址 nginx1 192.168.10.41 tomcat1 192.168.10.51 tomcat2 192.168.10.52 2、tomcat1 的配置 创建动态页面 cd /usr/local/tomcat/webapps 创建一个目录作为一个ser…

【LeetCode 链表合集】

文章目录 1. LeetCode 206 反转链表2. NC40 链表相加 1. LeetCode 206 反转链表 题目链接&#x1f517; 解题思路&#xff1a; &#x1f50d; &#x1f427;创建一个新的节点&#xff0c;使用链表头插的方法&#xff1b; 2. NC40 链表相加 题目链接&#x1f517; 解题思路…

C++入门基础(2)

C入门基础&#xff08;2&#xff09; 1.缺省函数2.函数重载3.引用3.1 引用的概念和定义3.2 引用的特性3.3 引用的使用3.3.1引用的特性 4 .const引用5. 指针和引用的关系6.inline 1.缺省函数 • 缺省参数是声明或定义函数时为函数的参数指定⼀个缺省值。在调用该函数时&#xf…

【服务器】在Linux查看运行的Python程序,并找到特定的Python程序

在Linux查看运行的Python程序并找到特定的Python程序 写在最前面1. 使用ps命令查看所有Python进程查看详细信息 2. 使用pgrep命令查找Python进程ID 3. 使用top或htop命令使用top命令使用htop命令 4. 使用lsof命令查找Python进程打开的文件 5. 使用nvidia-smi命令查看GPU使用情况…

【接口自动化_06课_Pytest+Excel+Allure完整框架集成】

一、logging在接口自动化里的应用 1、设置日志的配置&#xff0c;并收集日志文件 日志的设置需要在pytest.ini文件里设置。这个里面尽量不要有中文 2、debug日志的打印 pytest.ini文件的开关一定得是true才能在控制台打印日志 import allure import pytest from P06_PytestFr…

使用 YUM 仓库和 NFS 共享存储的详细指南

使用 YUM 仓库和 NFS 共享存储的详细指南 文章目录 使用 YUM 仓库和 NFS 共享存储的详细指南一、YUM 仓库服务1.1 YUM 介绍1.2 YUM 源的提供方式1.2.1 配置本地源仓库1.2.2 配置 FTP 源1.2.3 配置 HTTP 源 1.3 网络源配置1.3.1 清华源1.3.2 163 源1.3.3 阿里云源 1.4 YUM 命令1…

IntelliJ IDEA自定义菜单(Menus)、任务栏(toolbars)详细教程

本示例是基于IDEA2024.1Ultimate版本的New UI模式下 一、自定义菜单 1、打开Settings&#xff0c;找到Menus and Toolbars 2、点击右边的Main Menu&#xff0c;点击号&#xff0c;选择Add Action 3、弹出Add Action弹窗&#xff0c;搜索或者选择你要添加的指令 二、自定义工具…

Linux命令更新-Vim 编辑器

简介 Vim 是 Linux 系统中常用的文本编辑器&#xff0c;功能强大、可扩展性强&#xff0c;支持多种编辑模式和操作命令&#xff0c;被广泛应用于程序开发、系统管理等领域。 1. Vim 命令模式 Vim 启动后默认进入命令模式&#xff0c;此时键盘输入的命令将用于控制编辑器本身&…

OpenCV 寻找棋盘格角点及绘制

目录 一、概念 二、代码 2.1实现步骤 2.2完整代码 三、实现效果 一、概念 寻找棋盘格角点&#xff08;Checkerboard Corners&#xff09;是计算机视觉中相机标定&#xff08;Camera Calibration&#xff09;过程的重要步骤。 OpenCV 提供了函数 cv2.findChessboardCorners…

LeetCode 441, 57, 79

目录 441. 排列硬币题目链接标签思路代码 57. 插入区间题目链接标签思路两个区间的情况对每个区间的处理最终的处理 代码 79. 单词搜索题目链接标签原理思路代码 优化思路代码 441. 排列硬币 题目链接 441. 排列硬币 标签 数学 二分查找 思路 由于本题所返回的 答案在区间…