Spark性能调优与故障处理之(5)Spark 数据倾斜优化_spark数据膨胀-程序员宅基地

技术标签: spark  Spark  大数据  

Spark 中的数据倾斜问题主要指 shuffle 过程中出现的数据倾斜问题,是由于不同的 key 对应的数据量不同导致的不同 task 所处理的数据量不同的问题。

例如,reduce 点一共要处理 100 万条数据,第一个和第二个 task 分别被分配到了 1 万条数据,计算 5 分钟内完成,第三个 task 分配到了 98 万数据,此时第三个task 可能需要 10 个小时完成,这使得整个 Spark 作业需要 10 个小时才能运行完成,这就是数据倾斜所带来的后果。

注意:要区分开数据倾斜与数据量过量这两种情况
(1)数据倾斜是指少数 task 被分配了绝大多数的数据,因此少数 task运行缓慢;
(2)数据过量是指所有 task 被分配的数据量都很大,相差不多,所有 task 都运行缓慢。

数据倾斜的表现:

(1)Spark 作业的大部分 task 都执行迅速,只有有限的几个 task 执行的非常慢,此时可能出现了数据倾斜,作业可以运行,但是运行得非常慢;
(2)Spark 作业的大部分 task 都执行迅速,但是有的 task 在运行过程中会突然报出 OOM,反复执行几次都在某一个 task 报出 OOM 错误,此时可能出现了数据倾斜,作业无法正常运行。

定位数据倾斜问题:

(1)查阅代码中的 shuffle 算子,例如 reduceByKey、countByKey、groupByKey、join 等算子,根据代码逻辑判断此处是否会出现数据倾斜;
(2)查看 Spark 作业的 log 文件,log 文件对于错误的记录会精确到代码的某一行,可以根据异常定位到的代码位置来明确错误发生在第几个 stage,对应的 shuffle算子是哪一个;

一、聚合原数据

即:避免 shuffle 过程,从根本上消除发生数据倾斜的可能

绝大多数情况下,Spark 作业的数据来源都是 Hive 表,这些 Hive 表基本都是经过 ETL 之后的昨天的数据。

为了避免数据倾斜,我们可以考虑避免 shuffle 过程,如果避免了 shuffle 过程,那么从根本上就消除了发生数据倾斜问题的可能。

如果 Spark 作业的数据来源于 Hive 表,那么可以先在 Hive 表中对数据进行聚合,例如按照 key 进行分组,将同一 key 对应的所有 value 用一种特殊的格式拼接到一个字符串里去,这样,一个 key 就只有一条数据了;之后,对一个 key 的所有 value进行处理时,只需要进行 map 操作即可,无需再进行任何的 shuffle 操作。通过上述方式就避免了执行 shuffle 操作,也就不可能会发生任何的数据倾斜问题。

对于 Hive 表中数据的操作,不一定是拼接成一个字符串,也可以是直接对 key的每一条数据进行累计计算。

二、过滤导致倾斜的 key

如果在 Spark 作业中允许丢弃某些数据,那么可以考虑将可能导致数据倾斜的key 进行过滤,滤除可能导致数据倾斜的 key 对应的数据,这样,在 Spark 作业中就不会发生数据倾斜了。

三、提高shuffle操作中的reduce并行度

当以上两种方法对于数据倾斜的处理没有很好的效果时,可以考虑提高shuffle 过程中的 reduce 端并行度,reduce 端并行度的提高就增加了 reduce 端 task的数量,那么每个 task 分配到的数据量就会相应减少,由此缓解数据倾斜问题。

3.1 reduce 端并行度的设置

在大部分的 shuffle算子中 ,都可以传入一个并行度的设置参数 ,比 如reduceByKey(500),这个参数会决定 shuffle 过程中 reduce 端的并行度,在进行 shuffle操作的时候,就会对应着创建指定数量的 reduce task。对于 Spark SQL 中的 shuffle类语句,比如 group by、join 等,需要设置一个参数,即 spark.sql.shuffle.partitions,该参数代表了 shuffle read task 的并行度,该值默认是 200,对于很多场景来说都有点过小。

增加 shuffle read task 的数量,可以让原本分配给一个 task 的多个 key 分配给多个 task,从而让每个 task 处理比原来更少的数据。举例来说,如果原本有 5 个 key,每个 key 对应 10 条数据,这 5 个 key 都是分配给一个 task 的,那么这个 task 就要处理 50 条数据。而增加了 shuffle read task 以后,每个 task 就分配到一个 key,即每个 task 就处理 10 条数据,那么自然每个 task 的执行时间都会变短了。

3.2 reduce 端并行度设置存在的缺陷

提高 reduce 端并行度并没有从根本上改变数据倾斜的本质和问题(方案一和方案二从根本上避免了数据倾斜的发生),只是尽可能地去缓解和减轻 shuffle reducetask 的数据压力,以及数据倾斜的问题,适用于有较多 key 对应的数据量都比较大的情况。

该方案通常无法彻底解决数据倾斜,因为如果出现一些极端情况,比如某个 key对应的数据量有 100 万,那么无论你的 task 数量增加到多少,这个对应着 100 万数据的 key 肯定还是会分配到一个 task 中去处理,因此注定还是会发生数据倾斜的。所以这种方案只能说是在发现数据倾斜时尝试使用的第一种手段,尝试去用最简单的方法缓解数据倾斜而已,或者是和其他方案结合起来使用。

在理想情况下,reduce 端并行度提升后,会在一定程度上减轻数据倾斜的问题,甚至基本消除数据倾斜;但是,在一些情况下,只会让原来由于数据倾斜而运行缓慢的 task 运行速度稍有提升,或者避免了某些 task 的 OOM 问题,但是,仍然运行缓慢,此时,要及时放弃方案三,开始尝试后面的方案。

四、使用随机 key 实现双重聚合

当使用了类似于 groupByKey、reduceByKey 这样的算子时,可以考虑使用随机key 实现双重聚合,如下图所示:
在这里插入图片描述
首先,通过 map 算子给每个数据的 key 添加随机数前缀,对 key 进行打散,将原先一样的 key 变成不一样的 key,然后进行第一次聚合,这样就可以让原本被一个 task 处理的数据分散到多个 task 上去做局部聚合;随后,去除掉每个 key 的前缀,再次进行聚合。

此方法对于由 groupByKey、reduceByKey 这类算子造成的数据倾斜由比较好的效果,仅仅适用于聚合类的 shuffle 操作,适用范围相对较窄。如果是 join 类的 shuffle操作,还得用其他的解决方案。

此方法也是前几种方案没有比较好的效果时要尝试的解决方案。

五、将reduce join转换为 map join

正常情况下,join 操作都会执行 shuffle 过程,并且执行的是 reduce join,也就是先将所有相同的 key 和对应的 value 汇聚到一个 reduce task 中,然后再进行 join。

普通 join 的过程如下图所示:
在这里插入图片描述
普通的 join 是会走 shuffle 过程的,而一旦 shuffle,就相当于会将相同 key 的数据拉取到一个 shuffle read task 中再进行 join,此时就是 reduce join。但是如果一个RDD 是比较小的,则可以采用广播小RDD全量数据+map 算子来实现与 join 同样的效果,也就是 map join,此时就不会发生 shuffle 操作,也就不会发生数据倾斜。

注意:RDD 是并不能进行广播的,只能将 RDD 内部的数据通过 collect 拉取到 Driver 内存然后再进行广播

1、核心思路:

不使用 join 算子进行连接操作,而使用 Broadcast 变量与 map 类算子实现 join操作,进而完全规避掉 shuffle 类的操作,彻底避免数据倾斜的发生和出现。将较小RDD 中的数据直接通过 collect 算子拉取到 Driver 端的内存中来,然后对其创建一个 Broadcast 变量;接着对另外一个 RDD 执行 map 类算子,在算子函数内,从Broadcast 变量中获取较小 RDD 的全量数据,与当前 RDD 的每一条数据按照连接key 进行比对,如果连接 key 相同的话,那么就将两个 RDD 的数据用你需要的方式连接起来。

根据上述思路,根本不会发生 shuffle 操作,从根本上杜绝了 join 操作可能导致的数据倾斜问题。

当 join 操作有数据倾斜问题并且其中一个 RDD 的数据量较小时,可以优先考虑这种方式,效果非常好。map join 的过程如下图所示:
在这里插入图片描述

2、不适用场景分析:

由于 Spark 的广播变量是在每个 Executor 中保存一个副本,如果两个 RDD 数据量都比较大,那么如果将一个数据量比较大的 RDD 做成广播变量,那么很有可能会造成内存溢出。

六、sample 采样对倾斜 key单独进行join

在 Spark 中,如果某个 RDD 只有一个 key,那么在 shuffle 过程中会默认将此key 对应的数据打散,由不同的 reduce 端 task 进行处理。

由单个key导致数据倾斜时,可有将发生数据倾斜的 key 单独提取出来,组成一个 RDD,然后用这个原本会导致倾斜的 key 组成的 RDD 根据其他 RDD 单独 join,此时,根据 Spark 的运行机制,此 RDD 中的数据会在 shuffle 阶段被分散到多个 task中去进行 join 操作。

倾斜 key 单独 join 的流程如下图所示:
在这里插入图片描述
1、适用场景分析:
对于 RDD 中的数据,可以将其转换为一个中间表,或者是直接使用 countByKey()的方式,看一个这个 RDD 中各个 key 对应的数据量,此时如果你发现整个 RDD 就一个 key 的数据量特别多,那么就可以考虑使用这种方法

当数据量非常大时,可以考虑使用 sample 采样获取 10%的数据,然后分析这10%的数据中哪个 key 可能会导致数据倾斜,然后将这个 key 对应的数据单独提取出来

2、不适用场景分析:
如果一个 RDD 中导致数据倾斜的 key 很多,那么此方案不适用。

七、使用随机数以及扩容进行 join

如果在进行 join 操作时,RDD 中有大量的 key 导致数据倾斜,那么进行分拆 key也没什么意义,此时就只能使用最后一种方案来解决问题了,对于 join 操作,我们可以考虑对其中一个 RDD 数据进行扩容,另一个 RDD 进行稀释后再 join

我们会将原先一样的 key 通过附加随机前缀变成不一样的 key,然后就可以将这些处理后的“不同 key”分散到多个 task 中去处理,而不是让一个 task 处理大量的相同 key。这一种方案是针对有大量倾斜 key 的情况,没法将部分 key 拆分出来进行单独处理,需要对整个 RDD 进行数据扩容,对内存资源要求很高。

1、核心思想:

选择一个 RDD,使用 flatMap 进行扩容,对每条数据的 key 添加数值前缀(1~N的数值),将一条数据映射为多条数据;(扩容

选择另外一个 RDD,进行 map 映射操作,每条数据的 key 都打上一个随机数作为前缀(1~N 的随机数);(稀释

将两个处理后的 RDD,进行 join 操作。

在这里插入图片描述
2、局限性:
如果两个 RDD 都很大,那么将 RDD 进行 N 倍的扩容显然行不通;使用扩容的方式只能缓解数据倾斜,不能彻底解决数据倾斜问题。

3、使用方案七对方案六进一步优化分析:
当 RDD 中有几个 key 导致数据倾斜时,方案六不再适用,而方案七又非常消耗资源,此时可以引入方案七的思想完善方案六:

(1)对包含少数几个数据量过大的 key 的那个 RDD,通过 sample 算子采样出一份样本来,然后统计一下每个 key 的数量,计算出来数据量最大的是哪几个 key。

(2) 然后将这几个 key 对应的数据从原来的 RDD 中拆分出来,形成一个单独的RDD,并给每个 key 都打上 n 以内的随机数作为前缀,而不会导致倾斜的大部分 key形成另外一个 RDD。

(3) 接着将需要 join 的另一个 RDD,也过滤出来那几个倾斜 key 对应的数据并形成一个单独的 RDD,将每条数据膨胀成 n 条数据,这 n 条数据都按顺序附加一个0~n 的前缀,不会导致倾斜的大部分 key 也形成另外一个 RDD。

(4)再将附加了随机前缀的独立 RDD 与另一个膨胀 n 倍的独立 RDD 进行 join,此时就可以将原先相同的 key 打散成 n 份,分散到多个 task 中去进行 join 了。

(5) 而另外两个普通的 RDD 就照常 join 即可。

(6) 最后将两次 join 的结果使用 union 算子合并起来即可,就是最终的 join 结果。

版权声明:本文为博主原创文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接和本声明。
本文链接:https://blog.csdn.net/weixin_43520450/article/details/108651456

智能推荐

PAT基础编程题目集 7-31 字符串循环左移-程序员宅基地

文章浏览阅读67次。输入一个字符串和一个非负整数N,要求将字符串循环左移N次。

微服务 架构 单体架构_kolla 微服务-程序员宅基地

文章浏览阅读173次。RightScale2017 年的调查报告 [35] 显示,超过 84% 的企业,DevOps 的采用率达到 78%,而众多的 DevOps 工具中,Docker 的使用率也达到了 35%,Kubernetes 的使用率也从 2016 年的 7%增长到 2017 年的 14%。据 RightScale2017 年的调查报告 [35],在云服务商提供的服务中,Docker 的使用占据了很大的比重,包括 AWS ECS(35%)、Azure 容器服务(11%)以及谷歌容器引擎(8%)。_kolla 微服务

无人机三维建模的优势有哪些?-程序员宅基地

文章浏览阅读1.2k次。无人机三维建模技术凭借机动灵活、快速高效、成本较低等优势,已成为测绘行业的重要角色。将技术应用到无人机上,其实就是做一个三维模型,建立起来的三维模型会更加真实和直观。_无人机三维建模

汽车行业大趋势——软件定义汽车_软件定义汽车 趋势-程序员宅基地

文章浏览阅读728次,点赞2次,收藏5次。促使传统整车厂加速转型布局车载软件领域,软件定义汽车时代加速到来!_软件定义汽车 趋势

CSS 中哪些属性可继承,哪些不可以?_css中哪些属性可以继承?-程序员宅基地

文章浏览阅读1.2k次,点赞4次,收藏6次。能继承的属性 1. 字体系列属性:font、font-family、font-weight、font-size、font-style; 2. 文本系列属性: 2.1)内联元素:color、line-height、word-spacing、letter-spacing、text-transform; 2.2)块级元素:text-indent、text-align; 3. 元素可见性:visibility 4. 表格布局属性:caption-side、border-collapse、border-s_css中哪些属性可以继承?

sublime配置python-rpel_Sublime Text3配置SublimeREPL快捷键的方法(Python)-程序员宅基地

文章浏览阅读41次。[{"caption": "SublimeREPL: Python","command": "run_existing_window_command", "args":{"id": "repl_python","file": "config/Python/Main.sublime-menu"}},{"caption": "SublimeREPL: Python - PDB current file..._sublime3配置python的sublimerepl的快捷键

随便推点

idea生成有参构造的快捷键_Idea_学习_09_Idea 方法自动生成参数默认名-程序员宅基地

文章浏览阅读672次。关于SQL递归查询在不同数据库中的实现方法比如表结构数据如下: Table:Tree ID Name ParentId 1 一级 0 2 二级 1 3 三级 2 4 四级 3 SQL SERVER 2005查询方法: //上查 with ...Gulp-前端进阶A-1毕业到转行以来有一年时间了,成为一名程序猿也有大半年了,之前在新浪上随便写写简单的学习过程,感觉不够像那么回事,现在接触前端也..._ide怎么快速构建形参

H265框架编码流程(一),安卓高级开发面试题_安卓h26编码-程序员宅基地

文章浏览阅读126次。RQT(Residual Quad-tree Transform)技术是一种基于四叉树结构的自适应变换技术,它为最优TU模式选择提供了很高的灵活性。大块的TU模式能够将能量更好地集中,小块的TU模式能够保存更多的图像细节。根据当前CU内残差特性,自适应选择变换块大小,可以在能量集中和细节保留两者做最优的折中,与传统的固定块大小的变换相比,RQT对编码效率贡献更大。变换编码和量化模块从原理上属于两个相互独立的过程,但是在HEVC中,两个过程相互结合,减少了计算的复杂度。4.环路滤波在HEVC中,环路滤波_安卓h26编码

机器学习实战——疫情数据分析与预测_新冠疫情数据预期分析-程序员宅基地

文章浏览阅读1.8w次,点赞192次,收藏452次。机器学习如何做到疫情可视化——疫情数据分析与预测实战本文将带领大家爬取11个国家以及中国31个省(自治区、直辖市)在2022.0101-2022.06.19的新冠疫情数据。并且采用机器学习模型对2022.6.20-2022.6.30每一天的全国确诊人数、死亡人数、治愈人数进行预测,**做出疫情可视化图形并且求出最终的相关系数R2!..._新冠疫情数据预期分析

Axure的基本功能_axure主要功能-程序员宅基地

文章浏览阅读1.1k次。Axure的基本功能_axure主要功能

永磁同步电机pmsm无感foc驱动代码,启动为高频注入_foc高频注入 启动角-程序员宅基地

文章浏览阅读3.2k次。永磁同步电机pmsm无感foc驱动代码,启动为高频注入,平滑切入观测器高速控制,代码全部手写开源,可以移植到各类mcu上。附赠高频注入仿真模型现有:84100646985514964景龙池高级的玉竹_foc高频注入 启动角

openshift_OpenShift v3:使用WildFly和MySQL的Java EE 7入门-程序员宅基地

文章浏览阅读111次。openshift OpenShift是Red Hat的开源PaaS平台。 OpenShift v3 (将于今年发布)将提供使用Docker和Kubernetes运行微服务的整体体验。 以经典的Red Hat方式,所有工作都在OpenShift Origin的开源中完成。 这也将推动OpenShift Online和OpenShift Enterprise的下一个主要版本。 OpenShi..._openshift online v3