Spark九:Spark调优之Shuffle调优

这篇具有很好参考价值的文章主要介绍了Spark九:Spark调优之Shuffle调优。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

Spark shuffle调优方法

map端和reduce端缓存大小设置,reduce端重试次数和等待时间间隔,以及bypass设置
学习资料:https://mp.weixin.qq.com/s/caCk3mM5iXy0FaXCLkDwYQ

一、map和reduce端缓冲区大小

1.1 map端

在Spark任务运行过程中,如果shuffle的map端处理的数据量比较大,但是map端缓冲的大小是固定的,可能会出现map端缓冲数据频繁spill溢写到磁盘文件中的情况,使得性能非常低下。
通过调节map端缓冲的大小,可以避免频繁的磁盘IO操作,进而提升Spark任务的整体性能。


map端缓冲的默认配置是32KB,如果每个task处理640kb数据,那么会发生640/32=20次溢写,如果每个task处理64000KB数据,则发生2000次溢写,这对于性能的影响是非常重要的。
map端缓冲的配置方法

val conf = new SparkConf().set("spark.shuffle.file.buffer", "64")

1.2 reduce端

Spark Shuffle过程中,shuffle reduce task的buffer缓冲区大小决定了reduce task每次能够缓冲的数据量,也就是每次能够拉取的数据量,如果内存资源较为充足,适当增加拉取数据缓冲区的大小,可以减少拉取数据的次数,也就可以减少网络传输的次数,进而提升性能


reduce端数据拉去缓冲区的大小可以通过spark.reducer.maxSizeInFlight设置,默认为48M,设置方法:

val conf = new SparkConf().set("spark.reducer.maxSizeInFlight", "96")

二、reduce端重试次数和等待时间间隔

2.1 重试次数

Spark Shuffle过程中,reduce task拉取属于自己的数据时,如果因为网络异常等原因导致失败会自动进行重试。对于那些包含了特别耗时的shuffle操作的作业,建议增加重试最大次数(比如60次),以避免由于JVM的full gc或者网络不稳定等因素导致的数据拉取失败。在实践中发现,对于针对超大数据量(数十亿~上百亿)的shuffle过程,调节该参数可以大幅度提升稳定性。


reduce端拉取数据重试次数可以通过spark.shuffle.io.maxRetries参数设置,该参数就代表了可以重试的最大次数。如果在指定次数之内拉取还是没有成功,就可能会导致作业执行失败,默认为3,该参数的设置方法如下:

val conf = new SparkConf().set("spark.shuffle.io.maxRetries", "6")

2.2 增大等待时间间隔

Spark Shuffle过程中,reduce task拉取属于自己的数据时,如果因为网络异常等原因导致失败会自动进行重试,在一次失败后,会等待一定的时间间隔再进行重试,可以通过加大间隔时长(比如60s),以增加shuffle操作的稳定性
reduce端拉取数据等待间隔可以通过spark.shuffle.io.retryWait参数进行设置,默认值为5s,该参数的设置方法如下:

val conf = new SparkConf().set("spark.shuffle.io.retryWait", "60s")

三、bypass机制开启阈值

对于SortShuffleManager,如果shuffle reduce task的数量小于某一阈值,则shuffle write过程中不会进行排序操作,而是直接按照未经优化的HashShuffleManager的方式去写数据,但是最后会将每个task产生的所有临时磁盘文件都合并成一个文件,并会创建单独的索引文件。

当使用SortShuffleManager且不需要排序操作,建议将SortShuffleManager参数调大,大于shuffle read task的数量,那么此时map-side就不会进行排序了,减少了排序的性能开销,但是这种方式下,依然会产生大量的磁盘文件,因此shuffle write性能有待提高。


可以通过spark.shuffle.sort.bypassMergeThreshold这个参数设置,默认200。文章来源地址https://www.toymoban.com/news/detail-806494.html

val conf = new SparkConf().set("spark.shuffle.sort.bypassMergeThreshold", "400")

到了这里,关于Spark九:Spark调优之Shuffle调优的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • Spark Shuffle 过程

    本篇主要阐述Spark Shuffle过程,在执行 Job 任务时,无论是 MapReduce 或者 Spark Shuffle 过程都是比较消耗性能;因为该环节包含了大量的磁盘 IO、序列化、网络数据传输等操作。因此,在这一过程中进行调参优化,就有可能让 Job 执行效率上更好。 在 Spark 1.2 以前,默认的 Shuffle

    2023年04月08日
    浏览(38)
  • Spark学习笔记【shuffle】

    本文基本上是大数据处理框架Apache Spark设计与实现的Shuffle部分的学习。以及Spark基础知识@Bambrow 上游和下游,不同stage,不同的task之间是如何传递数据的。 ShuffleManager 管理ShuffleWrite和ShuffleRead 分为两个阶段: ShuffleWrite 上游stage输出的分区问题。 ShuffleRead 下游stage从上游获取

    2024年02月03日
    浏览(60)
  • Spark 读写 es 数据(scala 版)

    读取 hdfs 文件 解析采用 fast-json : 1、 pom.xml 2、 main 文件 运行结果: 1、 pom.xml 2、 main 文件 参考文章 Spark读写ES数据时遇到的问题总结 Spark读写ES 使用Apache Spark将数据写入ElasticSearch

    2024年02月11日
    浏览(42)
  • Spark Shuffle Tracking 原理分析(1)

    如果开启了 ESS,那么 Executor 计算完后,把 shuffle 数据交给 ESS, Executor 没有任务时,可以安全退出,下游任务从 ESS 拉取 shuffle 数据。 1. 背景 如果 Executor 执行了上游的 Shuffle Map Task 并且把 shuffle 数据些到本地。并且现在 Executor 没有 Task 运行,那么此 Executor 是否能销毁? 现状

    2024年04月16日
    浏览(27)
  • 简单使用Spark、Scala完成对天气数据的指标统计

    目录 一、前言   什么是Spark?   什么是Scala 二、数据准备(数据类型的转换) 三、Spark部分 1、使用Spark完成数据中的“风级”,“风向”、“天气情况”相关指标统计及筛选 四、Scala部分 1、使用Scala统计某月、全年的温差、平均气温以及最值等相关的指标 五、遇到的问题

    2024年02月03日
    浏览(50)
  • spark 数据序列化和内存调优(翻译)

    由于大多数Spark计算的内存性质,Spark程序可能会被集群中的任何资源瓶颈:CPU、网络带宽或内存。大多数情况下,如果数据能放在内存,瓶颈是网络带宽,但有时,您还需要进行一些调整,例如以序列化形式存储RDD,以减少内存使用。本指南将涵盖两个主要主题:数据序列化

    2024年03月11日
    浏览(51)
  • Spark 提交任务参数设置关于(线程,shuffle,序列化)

    是在使用 Apache Spark 时,为了设置 Java 虚拟机(JVM)的堆栈大小而使用命令行选项。 -Xss 是 Java 虚拟机的一个选项,用于设置线程的堆栈大小。在这个命令行选项中, -Xss6m 表示将线程的堆栈大小设为 6MB。这个选项的作用是为了避免在运行 Spark 任务时出现堆栈溢出的错误。

    2024年02月02日
    浏览(570)
  • 大数据Spark SparkSession的3种创建方式 Scala语言实现

    SparkSession是Apache Spark 2.0版本引入的一个编程接口,用于与Spark进行交互。它是Spark应用程序的入口点,提供了一种方便的方式来创建DataFrame、DataSet和SQLContext等数据结构,并且可以配置各种Spark应用程序的选项。SparkSession还管理了Spark应用程序的运行环境,包括Spark集群的连接,

    2023年04月20日
    浏览(38)
  • spark 的group by ,join数据倾斜调优

    spark任务中最常见的耗时原因就是数据分布不均匀,从而导致有些task运行时间很长,长尾效应导致的整个job运行耗时很长 首先我们要定位数据倾斜,我们可以通过在spark ui界面中查看某个stage下的task的耗时,如果发现某些task耗时很长,对应要处理的数据很多,证明有数据倾斜

    2024年02月21日
    浏览(46)
  • 字节跳动 Spark Shuffle 大规模云原生化演进实践

    Spark 是字节跳动内部使用广泛的计算引擎,已广泛应用于各种大规模数据处理、机器学习和大数据场景。目前中国区域内每天的任务数已经超过 150 万,每天的 Shuffle 读写数据量超过 500 PB。同时某些单个任务的 Shuffle 数据能够达到数百 TB 级别。 与此同时作业量与 Shuffle 的数

    2024年02月04日
    浏览(41)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包