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方法等)来监控作业的运行状态和性能瓶颈,并根据监控结果进行调优。

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

如若内容造成侵权/违法违规/事实不符,请联系我们进行投诉反馈qq邮箱809451989@qq.com,一经查实,立即删除!

相关文章

实现Android夜间模式主题:从入门到精通

实现Android夜间模式主题:从入门到精通 随着用户对夜间模式的需求越来越高,Android开发者需要掌握如何在应用中实现夜间模式。本文将详细介绍在Android中实现夜间模式的步骤,包括配置、实现、以及一些最佳实践,帮助开发者创建更具吸引力和用户友好的应用。 夜间模式的优势…

Git分支结构

目录 1. 线性分支结构 2. 分叉与合并结构 3. 分支与标签的关系 4. 并行开发与分支管理策略 测试&#xff08;本机系统为Rocky_linux9.4&#xff09; 合并失败解决 删除分支 删除本地分支 删除远程分支 Git 中的分支结构是版本控制中非常重要的概念之一&#xff0c;它描…

基于Java+SpringMvc+Vue技术的就医管理系统设计与实现系统(源码+LW+部署讲解)

目录 界面展示 第六章 部分代码实现 6.1 Spring boot 配置代码 6.2 用户管理及登录登出代码 6.3 Md5 加密算法代码 6.4 部分数据库代码 六、论文参考&#xff1a; 七、其他案例&#xff1a; 系统介绍&#xff1a; 就医管理系统&#xff0c;也称为医院管理系统&#…

14.爬虫---Selenium 经典动态渲染工具的使用

14.Selenium 经典动态渲染工具的使用 1.查看chrome浏览器版本2.ChromeDriver 安装3.Selenium 安装4.验证安装5.基本用法5.1启动浏览器5.2导航到页面5.3查找元素5.3.1单个元素 find_element5.3.2多个元素 find_elements 5.4 执行操作5.5 动作链ActionChains5.6 执行 JavaScript …

LabVIEW平台从离散光子到连续光子的光子计数技术

光子计数技术用于将输入光子数转换为离散脉冲。常见的光子计数器假设光子是离散到达的&#xff0c;记录到来的每一个光子。但是&#xff0c;当两个或多个光子同时到达时&#xff0c;计数器会将其记录为单个脉冲&#xff0c;从而只计数一次。当连续光子到达时&#xff0c;离散光…

微软Edge浏览器全解析:从速度到安全性的全面体验

微软Edge浏览器&#xff0c;自2015年首次亮相以来&#xff0c;已经成为了浏览器市场上不可忽视的一股力量。它不仅集成了Windows 10的许多原生功能&#xff0c;还在速度和安全性上进行了大量的优化。本文将全面解析微软Edge浏览器的各项特性&#xff0c;带您领略这款浏览器的魅…

M12电源分配器:高效率与可靠性的工业解决方案

关键词&#xff1a;M12电源分配器、M12电源接线盒、M12电源分线盒 摘要 M12电源分配器是工业环境中实现高效与可靠电源管理的现代化解决方案。该设备采用坚固的金属外壳和封闭式设计&#xff0c;保证了在严苛工业条件下的稳定运行&#xff0c;达到IP67等级的防护性能&#xf…

合合信息“大模型加速器”亮相2024世界人工智能大会

文章目录 &#x1f4d1;引言一、大模型发展的挑战数据稀缺问题 二、大模型“加速器”解决方案概述文档解析引擎的特征 三、文档解析引擎的优势3.1 高速处理能力3.2 智能理解文档结构3.3 多种数据类型支持3.4 高精度数据提取3.5 应用广泛&#xff0c;适应性强 四、复杂图表解析4…

二分查找模板及例题

文章目录 模板一&#xff1a;使用场景&#xff1a;解释&#xff1a;例题&#xff1a;数的范围题意&#xff1a; 代码&#xff1a; 模板二&#xff1a;使用场景&#xff1a;解释&#xff1a;例题&#xff1a;[Building an Aquarium](https://codeforces.com/problemset/problem/…

龙蜥Anolis OS基于开源项目制作openssh 9.8p1 rpm包 —— 筑梦之路

环境信息 制作过程和centos 7几乎没有区别&#xff0c;此处就不再赘述。 CentOS 7基于开源项目制作openssh9.8p1 rpm二进制包修复安全漏洞CVE-2024-6387 —— 筑梦之路_cve-2024-6387修复-CSDN博客 制作成果展示 tree RPMS/ RPMS/ └── x86_64├── openssh-9.8p1-1.an7.…

Python32 极限学习机ELM

极限学习机&#xff08;ELM&#xff09;是一种简单的单层前馈神经网络&#xff08;SLFN&#xff09;学习算法。理论上&#xff0c;极限学习机算法&#xff08;ELM&#xff09;往往以极快的学习速度提供良好的性能&#xff08;属于机器学习算法&#xff09;&#xff0c;由Huang等…

Three.js相机简明教程

相机校准是 3D 计算机图形学中的一个基本概念&#xff0c;涉及设置虚拟相机以模拟真实世界相机的视角和行为。在 Three.js&#xff08;一种流行的 3D 渲染 JavaScript 库&#xff09;中&#xff0c;了解相机校准对于创建逼真且身临其境的 3D 场景至关重要。在本文中&#xff0c…

CinemachineBrain的属性简介

CinemachineBrain的属性简介 CinemachineBrain是Unity Cinemachine的核心组件&#xff0c;它和Camera组件挂载在一起&#xff0c;监控场景中所有的virtual camera。CinemachineBrain在inspector中暴露的属性如下&#xff1a; Live Camera和Live Blend分别表示当前active的virtu…

人工智能算法工程师(中级)课程6-sklearn机器学习之聚类问题与代码详解

大家好&#xff0c;我是微学AI&#xff0c;今天给大家介绍一下人工智能算法工程师(中级)课程6-sklearn机器学习之聚类问题与代码详解。在机器学习领域&#xff0c;聚类是一种无监督学习方法&#xff0c;旨在将相似的数据点划分为同一类别。sklearn是一个广泛应用于机器学习的Py…

第十八章 Express multer 文件上传

本章将学习Express multer 文件上传 &#xff0c;因为Nest 的文件上传是基于 Express 的中间件 multer 实现的&#xff0c;所以在学习 Nest 文件上传之前&#xff0c;我们先学习下 multer 包 首先先创建 multer-test 文件夹执行下面代码 创建package.json npm init -y接着安装…

@RequiredArgsConstructor实现构造器注入

RequiredArgsConstructor实现构造器注入 1. Autowired 和 Resource 注解 Autowired Autowired 是 Spring 框架提供的注解&#xff0c;用于自动装配依赖。可以用于字段、构造函数和 setter 方法。 Autowired private ISysUserService userService;Resource Resource 是 Jav…

Java 中的 switch 语句:类型支持与限制

Java 中的 switch 语句&#xff1a;类型支持与限制 1、switch 语句支持的数据类型2、switch 语句不支持的数据类型3、总结 &#x1f496;The Begin&#x1f496;点点关注&#xff0c;收藏不迷路&#x1f496; 在 Java 中&#xff0c;switch 语句是一种用于多分支选择的控制结构…

物联网专业现代学徒制人才培养质量评价体系构建

一、 引 言 随着信息技术的飞速发展&#xff0c;物联网&#xff08;IoT&#xff09;技术已成为推动全球信息化、智能化发展的关键力量。物联网专业人才的培养质量直接关系到行业的创新能力和竞争力。现代学徒制作为一种创新的人才培养模式&#xff0c;已被广泛应用于职业教育中…

HCIP.ppp协议(点到点)认证阶段

ppp协议 ppp是点到点的协议 1.兼容性很好 2.可以进行认证和授权 3.可移植性强 三个阶段 1.链路协商阶段 LCP协商------去协商ppp链路会话 2.认证&#xff08;可选&#xff09; 3.NCP协商------网络层协商阶段&#xff08;根据网络层的不同NCP协议就会存在一个对应的NC…

查看尝试登服务器ssh 访问ip地址

不指定时间查看尝试登录服务器的SSH访问IP地址 # CentOS/RHEL系统 zgrep "sshd" /var/log/secure-* | grep "Failed password" | awk {print $(NF-3)} | sort | uniq -c | sort -nr | head -n 10检查过去7天的日志尝试登录服务器的SSH访问IP地址 # CentOS…