MapReduce性能优化--数据倾斜问题

深藏阁楼爱情的钟 2022-08-30 01:44 191阅读 0赞

我们来分析一个场景:  
假设我们有一个文件，有1000W条数据，这里面的值主要都是数字，1,2,3,4,5,6,7,8,9,10，我们希望统计出来每个数字出现的次数

其实在私底下我们是知道这份数据的大致情况的，这里面这1000w条数据，值为5的数据有910w条左右，剩下的9个数字一共只有90w条，那也就意味着这份数据中，值为5的数据比较集中，或者说值为5的数据属于`倾斜数据`，在这一整份数据中，它占得比重比其他的数据多得多。

假设这1000W条数据的文件有3个block，会产生3个InputSplt，最终会产生3个Map任务，默认情况下只有一个reduce任务，所以所有的数据都会让这一个reduce任务处理，这样这个Reduce压力肯定很大，大量的时间都消耗在了这里

那根据我们前面的分析，我们可以增加reduce任务的数量，我们把reduce任务的数量调整到10个，这个时候就会把1000w条数据让这10 个reduce任务并行处理了，这个时候效率肯定会有一定的提升，但是最后我们会发现，性能提升是有限的，并没有达到质的提升，那这是为什么呢?

我们来分析一下，刚才我们说了我们这份数据中，值为5的数据有910w条，这就占了整份数据的90% 了，那这90%的数据会被一个reduce任务处理，在这里假设是让reduce5处理了，reduce5这个任务执行 的是比较慢的，其他reduce任务都执行结束很长时间了，它还没执行结束，因为reduce5中处理的数据 量和其他reduce中处理的数据量规模相差太大了，所以最终reduce5拖了后腿。咱们mapreduce任务执 行消耗的时间是一直统计到最后一个执行结束的reduce任务，所以就算其他reduce任务早都执行结束了 也没有用，整个mapreduce任务是没有执行结束的。

那针对这种情况怎么办?  
这个时候单纯的增加reduce任务的个数已经不起多大作用了，如果启动太多可能还会适得其反。 其实这个时候最好的办法是把这个值为5的数据尽量打散，把这个倾斜的数据分配到其他reduce任务中去 计算，这样才能从根本上解决问题。

这就是我们要分析的一个数据倾斜的问题  
MapReduce程序执行时，Reduce节点大部分执行完毕，但是有一个或者几个Reduce节点运行很慢，导 致整个程序处理时间变得很长  
具体表现为:Ruduce阶段一直卡着不动

根据刚才的分析，有两种方案：

1.  **增加reduce任务个数**：这个属于治标不治本，针对倾斜不是太严重的数据是可以解决问题的，针对倾斜严重的数据，这样是解决不了根本问题的
2.  **把倾斜的数据打散**：这种可以根治倾斜严重的数据。

下面我们主要讲方案2的实现方式  
其实思路很简单，在这里我们这样处理，把5这个数值的数据再分成10份，就在这个数值5后面拼上一个0~9的随机数即可。

import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Job;
    import org.apache.hadoop.mapreduce.Mapper;
    import org.apache.hadoop.mapreduce.Reducer;
    import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import java.io.IOException;
    import java.util.Random;
    
    public class WordCountJobSkewRandKey { 
        /** * Map阶段 */
        public static class MyMapper extends Mapper<LongWritable, Text,Text,LongWritable>{ 
            Logger logger = LoggerFactory.getLogger(MyMapper.class);
            Random random = new Random();
            
            /** * 需要实现map函数 * 这个map函数就是可以接收<k1,v1>，产生<k2，v2> */
            @Override
            protected void map(LongWritable k1, Text v1, Context context) throws IOException, InterruptedException {  
    	        // 输出k1,v1的值
    			System.out.println("<k1,v1>=<"+k1.get()+","+v1.toString()+">");
    			logger.info("<k1,v1>=<"+k1.get()+","+v1.toString()+">");
    			// k1 代表的是每一行数据的行首偏移量，v1代表的是每一行内容 
    			// 对获取到的每一行数据进行切割，把单词切割出来
    			String[] words = v1.toString().split(" "); 
    			// 把单词封装成<k2,v2>的形式
    			String key = words[0]; if("5".equals(key)){ 
    			//把倾斜的key打散，分成10份
    	                key = "5"+"_"+random.nextInt(10);
    	            }
    			Text k2 = new Text(key);
    			LongWritable v2 = new LongWritable(1L); //把<k2,v2>写出去
    			context.write(k2,v2);
    		}
    	}
    	
    	/** * Reduce阶段 */
        public static class MyReducer extends Reducer<Text,LongWritable,Text,LongWritable>{ 
            Logger logger = LoggerFactory.getLogger(MyReducer.class);
            
            /** * 针对<k2,{v2...}>的数据进行累加求和，并且最终把数据转化为k3,v3写出去 */
            @Override
            protected void reduce(Text k2, Iterable<LongWritable> v2s, Context co
    throws IOException, InterruptedException {  
    			// 创建一个sum变量，保存v2s的和
    			long sum = 0L;
    			// 对v2s中的数据进行累加求和
    			for(LongWritable v2: v2s){  
    				// 输出k2,v2的值
    				//System.out.println("<k2,v2>=<"+k2.toString()+","+v2.get()+"
                    //logger.info("<k2,v2>=<"+k2.toString()+","+v2.get()+">");
                    sum += v2.get();
    				//模拟Reduce的复杂计算消耗的时间 
    				if(sum % 200 ==0){ 
                        Thread.sleep(1);
                    }
    			}
    			
    			// 组装k3,v3
    			Text k3 = k2;
    			LongWritable v3 = new LongWritable(sum);
    			// 输出k3,v3的值 
    			//System.out.println("<k3,v3>=<"+k3.toString()+","+v3.get()+">");
    			//logger.info("<k3,v3>=<"+k3.toString()+","+v3.get()+">");
    			// 把结果写出去
    			context.write(k3,v3);
    		}
    	}
    
    	/** * 组装Job=Map+Reduce */
    	public static void main(String[] args) { 
    		try { 
    			if(args.length!=3){  
    				//如果传递的参数不够，程序直接退出 
    				System.exit(100);
    			}
    			
    			//指定Job需要的配置参数
    			Configuration conf = new Configuration(); 
    			//创建一个Job
    			Job job = Job.getInstance(conf);
    			
    			job.setJarByClass(WordCountJobSkew.class);
    			// 指定输入路径(可以是文件，也可以是目录) 
    			FileInputFormat.setInputPaths(job,new Path(args[0])); 
    			// 指定输出路径(只能指定一个不存在的目录) 
    			FileOutputFormat.setOutputPath(job,new Path(args[1]));
    			// 指定map相关的代码 
    			job.setMapperClass(MyMapper.class); 
    			// 指定k2的类型 
    			job.setMapOutputKeyClass(Text.class); 
    			// 指定v2的类型
    			job.setMapOutputValueClass(LongWritable.class);
    			
    			// 指定reduce相关的代码
    			job.setReducerClass(MyReducer.class); 
    			// 指定k3的类型
    			job.setOutputKeyClass(Text.class);
    			// 指定v3的类型 
    			job.setOutputValueClass(LongWritable.class); 
    			// 设置reduce任务个数 
    			job.setNumReduceTasks(Integer.parseInt(args[2]));
    			
    			// 提交job
            	job.waitForCompletion(true);
        	} catch (Exception e){ 
        		e.printStackTrace();
       		}
    	}
    }

执行成功之后查看结果

[root@bigdata01 hadoop-3.2.0]# hdfs dfs -cat /out10000000/*
    1       100000
    5_3     1012097
    2       100000
    5_4     1011163
    3       100000
    5_5     1010498
    4       100000
    5_6 	1010755
    5_7 	1010823
    5_8 	1012394
    6 		100000
    7 		100000
    5_0     1011274
    8       100000
    10      100000
    5_1     1009972
    9       100000
    5_2     1011024

但是这个时候我们获取到的最终结果是一个半成品，还需要进行一次加工  
其实我们前面把这个倾斜的数据打散之后相当于做了一个局部聚合，现在还需要再开发一个mapreduce任务再做一次全局聚合，  
其实也很简单，获取到上一个map任务的输出，在map端读取到数据之后，对数据先使用空格分割，然后对第一列的数据再使用下划线分割，分割之后总是取第一列，这样就可以把值为5的数据还原出来了