zoukankan      html  css  js  c++  java
  • 最完整的数据倾斜解决方案(spark)

    一.了解数据倾斜

      数据倾斜的原理:

          

          在执行shuffle操作的时候,按照key,来进行values的数据的输出,拉取和聚合.同一个key的values,一定是分配到一个Reduce task进行处理.

          假如多个key对应的values,总共是90万,但是可能某个key对应了88万条数据,key-88万条values,分配到一个task上面去执行.

          另外两个task,可能各分配到了1万条数据,可能是数百个key,对应一万条数据.

      数据倾斜的现象:

        发生数据倾斜的两种表现: 

          1.你的大部分的task,都执行的特别特别快,在很短的时间就执行完了(用client模式,standalone client,yarn client,本地机器主要一执行spark-submit脚本,就会开始打印log),比如task88 finished,剩下几个task,执行的特别慢,前面的task,一般1s可以执行完5个,最后发现1000个task,998,999task,要执行1个小时,甚至两个小时才能执行完一个task.

          2.运行的时候,同样是其他的task都很快执行完了,也没什么特别的问题,但是有的task,就是会突然间,啪,报了一个OOM,JVM Out Of Memory,内存溢出了,task failed, task lost, resubmitting task.反复提交执行了几次都到了某个task就是跑不通,最后就会挂掉.某个task就直接OOM,那么基本上也是因为数据倾斜了,task分配的数量实在是太大了!!!所以内存放不下,然后你的task每处理一条数据,还要创建大量的对象。内存爆掉了。

      数据倾斜的产生原因与定位:

          根据log去定位:

          出现数据倾斜的原因,基本只可能是因为发生了shuffle操作,在shuffle的过程中,出现了数据倾斜的问题,因为某个,或者某些key对应的数据,远远高于其他的key.

          1.在所写的程序找找,哪些地方用到了回产生shuffle的算子,groupByKey、countByKey、reduceByKey、join,groupBy,repartition,cogroup,distinct,leftouterJoin

          2.看log,log一般会报是在你的哪一行代码,导致了OOM异常,或者呢,看log,看看是执行到了第几个stage(stage划分).

    二.解决数据倾斜的方案

      方案一:聚合源数据

           情况一:     

              (避免聚合)spark算子聚合作业,其实就是groupByKey、reduceByKey,其实就是拿到每个key对应的values;reduceByKey,其实就是对每个key对应的values执行一定的计算。

              这些有可能导致数据倾斜的操作,比如groupByKey和reduceByKey,包括之前说的join。都是在spark作业中执行的。

              spark作业的数据来源通常(90%)的情况下,数据来源都是hive表(hdfs,大数据分布式存储系统)。hdfs上存储的大数据。

              hive表,hive表中的数据,通常是怎么出来的呢?有了spark以后,hive比较适合做什么事情?hive就是适合做离线的,晚上凌晨跑的,ETL(extract transform load,数据的采集、清洗、导入),hive sql,去做这些事情,从而去形成一个完整的hive中的数据仓库;说白了,数据仓库,就是一堆表。

              spark作业的源表(hive表),其实通常情况下来说,也是通过某些hive etl生成的。hive etl可能是晚上凌晨在那儿跑。今天跑昨天的数据。

               数据倾斜,某个key对应的80万数据,某些key对应几百条,某些key对应几十条;现在我们直接在生成hive表的hive etl中,对数据进行聚合。比如按key来分组,将key对应的所有的values,全部用一种特殊的格式,拼接到一个字符串里面去,比如“key=sessionid, value: action_seq=1|user_id=1|search_keyword=火锅|category_id=001;action_seq=2|user_id=1|search_keyword=涮肉|category_id=001”。

               在hive的Hql中我们可以对key进行group,然后在spark中,拿到key=sessionid,values<Iterable>;hive etl中,已经对key进行了聚合。那么也就意味着,每个key就只对应一条数据。在spark中,就不需要再去执行groupByKey+map这种操作了。直接对每个key对应的values字符串,map操作,进行你需要的操作即可。key,values串。

              spark中,可能对这个操作,就不需要执行shffule操作了,也就根本不可能导致数据倾斜。 或者是,对每个key在hive etl中进行聚合,对所有values聚合一下,不一定是拼接起来,可能是直接进行计算。reduceByKey,计算函数,应用在hive etl中,每个key的values

              情况二:     

              (增粗聚合粒度)我们可能没有办法对每个key,就聚合出来一条数据; 那么也可以做一个妥协;对每个key对应的数据,10万条;有好几个粒度,比如10万条里面包含了几个城市、几天、几个地区的数据,现在放粗粒度;直接就按照城市粒度,做一下聚合,几个城市,几天、几个地区粒度的数据,都给聚合起来。

              比如说 city_id date area_id   select ... from ... group by city_id

               尽量去聚合,减少每个key对应的数量,也许聚合到比较粗的粒度之后,原先有10万数据量的key,现在只有1万数据量。减轻数据倾斜的现象和问题。

      方案二:过滤导致倾斜的key

           如果可以接受某些数据,在spark作业中直接就摒弃掉,不使用,比如说,总共有100万个key,只有2个key,是数据量达到10万的,其他所有的key,对应的数据量都是几十.

          这个时候,我们自己可以进行取舍,如果业务和需求可以理解和接受的话,在我们从hive表查询元数据的时候,直接在sql中用where条件,过滤掉某几个key.

          这个几个原先有大量数据,会导致数据倾斜的key,被过滤掉之后,在我们的Spark作业中,自然就不会发生数据倾斜了.

      方案三:提高shuffle操作的reduce并行度

          

          

          将reduce task的数量,变多,就可以让每个reduce task分配到更少的数据量,这样的话,也许就可以缓解,或者甚至是基本解决掉数据倾斜的问题。

          具体方法就是在shuffle算子后面指定task分区数,比如val rdd2 = rdd1.reduceByKey(_+_,10)      

          提升shuffle reduce端并行度,怎么来操作?

              很简单,主要给我们所有的shuffle算子,比如groupByKey、countByKey、reduceByKey。在调用的时候,传入进去一个参数。一个数字。那个数字,就代表了那个shuffle操作的reduce端的并行度。那么在进行shuffle操作的时候,就会对应着创建指定数量的reduce task。

              这样的话,就可以让每个reduce task分配到更少的数据。基本可以缓解数据倾斜的问题。

              比如说,原本某个task分配数据特别多,直接OOM,内存溢出了,程序没法运行,直接挂掉。按照log,找到发生数据倾斜的shuffle操作,给它传入一个并行度数字,这样的话,原先那个task分配到的数据,肯定会变少。就至少可以避免OOM的情况,程序至少是可以跑的。

          提升shuffle reduce并行度的缺陷:      

              治标不治本的意思,因为,它没有从根本上改变数据倾斜的本质和问题。不像第一个和第二个方案(直接避免了数据倾斜的发生)。原理没有改变,只是说,尽可能地去缓解和减轻shuffle reduce task的数据压力,以及数据倾斜的问题。 

               1、如果最理想的情况下,提升并行度以后,减轻了数据倾斜的问题,或者甚至可以让数据倾斜的现象忽略不计,那么就最好。就不用做其他的数据倾斜解决方案了。

              2、不太理想的情况下,就是比如之前某个task运行特别慢,要5个小时,现在稍微快了一点,变成了4个小时;或者是原先运行到某个task,直接OOM,现在至少不会OOM了,但是那个task运行特别慢,要5个小时才能跑完。

      方案四:使用随机key实现双重group聚合方案

    		/**
    		 * 使用随机key实现双重聚合
    		 */
    		
    		/**
    		 * 第一步,给每个key打上一个随机数
    		 */
    		JavaPairRDD<String, Long> mappedClickCategoryIdRDD = clickCategoryIdRDD.mapToPair(
    				
    				new PairFunction<Tuple2<Long,Long>, String, Long>() {
    
    					private static final long serialVersionUID = 1L;
    		
    					@Override
    					public Tuple2<String, Long> call(Tuple2<Long, Long> tuple)
    							throws Exception {
    						Random random = new Random();
    						int prefix = random.nextInt(10);
    						return new Tuple2<String, Long>(prefix + "_" + tuple._1, tuple._2);
    					}
    					
    				});
    		
    		/**
    		 * 第二步,执行第一轮局部聚合
    		 */
    		JavaPairRDD<String, Long> firstAggrRDD = mappedClickCategoryIdRDD.reduceByKey(
    				
    				new Function2<Long, Long, Long>() {
    
    					private static final long serialVersionUID = 1L;
    
    					@Override
    					public Long call(Long v1, Long v2) throws Exception {
    						return v1 + v2;
    					}
    					
    				});
    		
    		/**
    		 * 第三步,去除掉每个key的前缀
    		 */
    		JavaPairRDD<Long, Long> restoredRDD = firstAggrRDD.mapToPair(
    				
    				new PairFunction<Tuple2<String,Long>, Long, Long>() {
    
    					private static final long serialVersionUID = 1L;
    		
    					@Override
    					public Tuple2<Long, Long> call(Tuple2<String, Long> tuple)
    							throws Exception {
    						long categoryId = Long.valueOf(tuple._1.split("_")[1]);  
    						return new Tuple2<Long, Long>(categoryId, tuple._2);  
    					}
    					
    				});
    		
    		/**
    		 * 第四步,最第二轮全局的聚合
    		 */
    		JavaPairRDD<Long, Long> clickCategoryId2CountRDD = restoredRDD.reduceByKey(
    				
    				new Function2<Long, Long, Long>() {
    
    					private static final long serialVersionUID = 1L;
    
    					@Override
    					public Long call(Long v1, Long v2) throws Exception {
    						return v1 + v2;
    					}
    					
    				});
    

     

           1.将一个热点的key进行加盐(就是加上随机的前缀)

           2.然后对进行加盐处理后的key进行reduceByKey,groupByKey等算子操作

          3.去掉key的前缀.

          4.重复步骤2操作,避免了一个key对应的value过多的算子操作(数据倾斜).

      方案五:将reduce join 转换为map join

          原理图:

            /**
             * reduce join转换为map join
             */
            
            List<Tuple2<Long, Row>> userInfos = userid2InfoRDD.collect();
            final Broadcast<List<Tuple2<Long, Row>>> userInfosBroadcast = sc.broadcast(userInfos);
            
            JavaPairRDD<String, String> sessionid2FullAggrInfoRDD = userid2PartAggrInfoRDD.mapToPair(
                    
                    new PairFunction<Tuple2<Long,String>, String, String>() {
    
                        private static final long serialVersionUID = 1L;
    
                        @Override
                        public Tuple2<String, String> call(Tuple2<Long, String> tuple)
                                throws Exception {
                            // 得到用户信息map
                            List<Tuple2<Long, Row>> userInfos = userInfosBroadcast.value();
                            
                            Map<Long, Row> userInfoMap = new HashMap<Long, Row>();
                            for(Tuple2<Long, Row> userInfo : userInfos) {
                                userInfoMap.put(userInfo._1, userInfo._2);
                            }
                            
                            // 获取到当前用户对应的信息
                            String partAggrInfo = tuple._2;
                            Row userInfoRow = userInfoMap.get(tuple._1);
                            
                            String sessionid = StringUtils.getFieldFromConcatString(
                                    partAggrInfo, "\|", Constants.FIELD_SESSION_ID);
                            
                            int age = userInfoRow.getInt(3);
                            String professional = userInfoRow.getString(4);
                            String city = userInfoRow.getString(5);
                            String sex = userInfoRow.getString(6);
                            
                            String fullAggrInfo = partAggrInfo + "|"
                                    + Constants.FIELD_AGE + "=" + age + "|"
                                    + Constants.FIELD_PROFESSIONAL + "=" + professional + "|"
                                    + Constants.FIELD_CITY + "=" + city + "|"
                                    + Constants.FIELD_SEX + "=" + sex;
                            
                            return new Tuple2<String, String>(sessionid, fullAggrInfo);
                        }
                        
                    });

           reduce join转换map join,适合在什么样的情况下,可以来使用?

              如果两个Rdd要进行join,其中一个RDD是比较小的,一个RDD是100万条数据,一个RDD是1万数据.

              其中一个RDD必须是比较小的,broadcast出去的那个小RDD的数据以后,就会在每个executor的block manager中都驻留一份.要确保你的内存足够放那个小RDD的数据.

              这种方式下,根本不会发生shuffle操作,肯定也不会发生数据倾斜;从根本上杜绝了join操作可能导致的数据倾斜的问题;

              对于join中有数据倾斜的情况,大家尽量第一时间先考虑这种方式,效果非常好;如果某个RDD比较小的情况下。

          不适合的情况:    

              两个RDD都比较大,那么这个时候,你去将其中一个RDD做成broadcast,就很笨拙了。很可能导致内存不足。最终导致内存溢出,程序挂掉。 而且其中某些key(或者是某个key),还发生了数据倾斜;此时可以采用最后两种方式。

          总结:   

              对于join这种操作,不光是考虑数据倾斜的问题;即使是没有数据倾斜问题,也完全可以优先考虑,用我们讲的这种高级的reduce join转map join的技术,不要用普通的join,去通过shuffle,进行数据的join;完全可以通过简单的map,使用map join的方式,牺牲一点内存资源;在可行的情况下,优先这么使用。 不走shuffle,直接走map,是不是性能也会高很多?这是肯定的。

      方案六:sample采样倾斜key进行两次join

          

          当第五种方案不能解决,就是两个RDD都比较大的情况下的join操作,直接针对产生数据倾斜的Key的处理方案.

          方案思路:

            在要进行join操作的时候,随机采样出10%数据,使用countByKey计算出key的数量然后sortByKey(false)倒序取出第一个key,然后对取出来的key和普通的key分别进行join操作之后再进行join操作.

            这个方案的实现思路,跟大家解析一下:其实关键之处在于,将发生数据倾斜的key,单独拉出来,放到一个RDD中去;就用这个原本会倾斜的key RDD跟其他RDD,单独去join一下,这个时候,key对应的数据,可能就会分散到多个task中去进行join操作。 就不至于说是,这个key跟之前其他的key混合在一个RDD中时,肯定是会导致一个key对应的所有数据,都到一个task中去,就会导致数据倾斜。

          这种方案什么时候适合使用?

            优先对于join,肯定是希望能够采用上一个方案的,reduce join转换map join。两个RDD数据都比较大,那么就不要那么搞了。

            针对你的RDD的数据,你可以自己把它转换成一个中间表,或者是直接用countByKey()的方式,你可以看一下这个RDD各个key对应的数据量;

            此时如果你发现整个RDD就一个,或者少数几个key,是对应的数据量特别多;尽量建议,比如就是一个key对应的数据量特别多。 此时可以采用咱们的这种方案,单拉出来那个最多的key;单独进行join,尽可能地将key分散到各个task上去进行join操作。

          什么时候不适用呢?

             如果一个RDD中,导致数据倾斜的key,特别多;那么此时,最好还是不要这样了;还是使用我们最后一个方案,终极的join数据倾斜的解决方案。

           升级版做法思路:

            就是说,咱们单拉出来了,一个或者少数几个可能会产生数据倾斜的key,然后还可以进行更加优化的一个操作; 对于那个key,从另外一个要join的表中,也过滤出来一份数据,比如可能就只有一条数据。userid2infoRDD,一个userid key,就对应一条数据。 然后呢,采取对那个只有一条数据的RDD,进行flatMap操作,打上100个随机数,作为前缀,返回100条数据。 单独拉出来的可能产生数据倾斜的RDD,给每一条数据,都打上一个100以内的随机数,作为前缀。 再去进行join,是不是性能就更好了。肯定可以将数据进行打散,去进行join。join完以后,可以执行map操作,去将之前打上的随机数,给去掉,然后再和另外一个普通RDD join以后的结果,进行union操作。

          代码实现:

            https://www.cnblogs.com/gentle-awen/p/10144882.html

      方案七:使用随机数以及扩容表进行join

          

          方案思路:

            针对上面方案都没办法解决数据倾斜,只能使用这种.当采用随机数和扩容表进行join解决数据倾斜的时候,就代表着,你的之前的数据倾斜的解决方案,都没法使用。 这个方案是没办法彻底解决数据倾斜的,更多的,是一种对数据倾斜的缓解。

          方案步骤:

            1、选择一个RDD,要用flatMap,进行扩容,将每条数据,映射为多条数据,每个映射出来的数据,都带了一个n以内的随机数,通常来说,会选择10。

            2、将另外一个RDD,做普通的map映射操作,每条数据,都打上一个10以内的随机数。

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

          方案局限性:

            1、因为你的两个RDD都很大,所以你没有办法去将某一个RDD扩的特别大,一般咱们就是10倍。

            2、如果就是10倍的话,那么数据倾斜问题,的确是只能说是缓解和减轻,不能说彻底解决。

          方案代码:

            https://www.cnblogs.com/gentle-awen/p/10144893.html

     

  • 相关阅读:
    java编程基础--方法
    MySQL中使用LIMIT进行分页的方法
    Java编程基础--数据类型
    Java开发入门
    SpringBoot实战项目(十七)--使用拦截器实现系统日志功能
    SpringBoot实战项目(十六)--拦截器配置及登录拦截
    SpringBoot实战项目(十五)--修改密码及登录退出功能实现
    SpringBoot实战项目(十四)--登录功能之登录表单验证
    PHP setcookie 网络函数
    PHP mysqli_kill MySQLi 函数
  • 原文地址:https://www.cnblogs.com/gentle-awen/p/10141315.html
Copyright © 2011-2022 走看看