Hadoop基础-MapReduce的数据倾斜解决方案

时间：2022-04-20 15:32:38

　　　　　　　　　　　　　　　　　　　　　Hadoop基础-MapReduce的数据倾斜解决方案

　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　作者：尹正杰

版权声明：原创作品，谢绝转载！否则将追究法律责任。

一.数据倾斜简介

1>.什么是数据倾斜

　　答：大量数据涌入到某一节点，导致此节点负载过重，此时就产生了数据倾斜。

2>.处理数据倾斜的两种方案

　　第一：重新设计key；

　　第二：设计随机分区；

二.模拟数据倾斜

screw.txt 文件内容

1>.App端代码

 /*

 @author :yinzhengjie

 Blog:http://www.cnblogs.com/yinzhengjie/tag/Hadoop%E8%BF%9B%E9%98%B6%E4%B9%8B%E8%B7%AF/

 EMAIL:y1053419035@qq.com

 */

 package cn.org.yinzhengjie.srew;

 import org.apache.hadoop.conf.Configuration;

 import org.apache.hadoop.fs.FileSystem;

 import org.apache.hadoop.fs.Path;

 import org.apache.hadoop.io.IntWritable;

 import org.apache.hadoop.io.Text;

 import org.apache.hadoop.mapreduce.Job;

 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

 public class ScrewApp {

     public static void main(String[] args) throws Exception {

         //实例化一个Configuration，它会自动去加载本地的core-site.xml配置文件的fs.defaultFS属性。(该文件放在项目的resources目录即可。)

         Configuration conf = new Configuration();

         //将hdfs写入的路径定义在本地，需要修改默认为文件系统，这样就可以覆盖到之前在core-site.xml配置文件读取到的数据。

         conf.set("fs.defaultFS","file:///");

         //代码的入口点，初始化HDFS文件系统，此时我们需要把读取到的fs.defaultFS属性传给fs对象。

         FileSystem fs = FileSystem.get(conf);

         //创建一个任务对象job，别忘记把conf穿进去哟！

         Job job = Job.getInstance(conf);

         //给任务起个名字

         job.setJobName("WordCount");

         //指定main函数所在的类，也就是当前所在的类名

         job.setJarByClass(ScrewApp.class);

         //指定map的类名,这里指定咱们自定义的map程序即可

         job.setMapperClass(ScrewMapper.class);

         //指定reduce的类名，这里指定咱们自定义的reduce程序即可

         job.setReducerClass(ScrewReduce.class);

         //设置输出key的数据类型

         job.setOutputKeyClass(Text.class);

         //设置输出value的数据类型

         job.setOutputValueClass(IntWritable.class);

         Path localPath = new Path("D:\\10.Java\\IDE\\yhinzhengjieData\\MyHadoop\\MapReduce\\out");

         if (fs.exists(localPath)){

             fs.delete(localPath,true);

         }

         //设置输入路径，需要传递两个参数，即任务对象(job)以及输入路径

         FileInputFormat.addInputPath(job,new Path("D:\\10.Java\\IDE\\yhinzhengjieData\\MyHadoop\\MapReduce\\screw.txt"));

         //设置输出路径，需要传递两个参数，即任务对象(job)以及输出路径

         FileOutputFormat.setOutputPath(job,localPath);

         //设置Reduce的个数为2.

         job.setNumReduceTasks();

         //等待任务执行结束,将里面的值设置为true。

         job.waitForCompletion(true);

     }

 }

ScrewApp.java 文件内容

2>.Reduce端代码

 /*

 @author :yinzhengjie

 Blog:http://www.cnblogs.com/yinzhengjie/tag/Hadoop%E8%BF%9B%E9%98%B6%E4%B9%8B%E8%B7%AF/

 EMAIL:y1053419035@qq.com

 */

 package cn.org.yinzhengjie.srew;

 import org.apache.hadoop.io.IntWritable;

 import org.apache.hadoop.io.Text;

 import org.apache.hadoop.mapreduce.Reducer;

 import java.io.IOException;

 public class ScrewReduce extends Reducer<Text,IntWritable,Text,IntWritable> {

     @Override

     protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {

         int count = ;

         for (IntWritable value : values) {

             count += value.get();

         }

         context.write(key,new IntWritable(count));

     }

 }

ScrewReduce.java 文件内容

3>.Mapper端代码

 /*

 @author :yinzhengjie

 Blog:http://www.cnblogs.com/yinzhengjie/tag/Hadoop%E8%BF%9B%E9%98%B6%E4%B9%8B%E8%B7%AF/

 EMAIL:y1053419035@qq.com

 */

 package cn.org.yinzhengjie.srew;

 import org.apache.hadoop.io.IntWritable;

 import org.apache.hadoop.io.LongWritable;

 import org.apache.hadoop.io.Text;

 import org.apache.hadoop.mapreduce.Mapper;

 import java.io.IOException;

 public class ScrewMapper extends Mapper<LongWritable,Text,Text,IntWritable> {

     @Override

     protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

         String line = value.toString();

         String[] arr = line.split(" ");

         for (String word : arr) {

             context.write(new Text(word),new IntWritable());

         }

     }

 }

ScrewMapper.java 文件内容

　　执行以上代码，查看数据如下：

Hadoop基础-MapReduce的数据倾斜解决方案

三.解决数据倾斜方案之重新设计key

1>.具体代码如下

/*

@author :yinzhengjie

Blog:http://www.cnblogs.com/yinzhengjie/tag/Hadoop%E8%BF%9B%E9%98%B6%E4%B9%8B%E8%B7%AF/

EMAIL:y1053419035@qq.com

*/

package cn.org.yinzhengjie.srew;

import org.apache.hadoop.io.IntWritable;

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

import java.util.Random;

public class ScrewMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

    //定义一个reduce变量

    int reduces;

    //定义一个随机数生成器变量

    Random r;

    /**

     * setup方法是用于初始化值

     */

    @Override

    protected void setup(Context context) throws IOException, InterruptedException {

        //通过context.getNumReduceTasks()方法获取到用户配置的reduce个数。

        reduces = context.getNumReduceTasks();

        //生成一个随机数生成器

        r = new Random();

    }

    @Override

    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

        String line = value.toString();

        String[] arr = line.split(" ");

        for (String word : arr) {

            //从reducs的范围中获取一个int类型的随机数赋值给randVal

            int randVal = r.nextInt(reduces);

            //重新定义key

            String newWord = word+"_"+ randVal;

            //将自定义的key赋初始值为1发给reduce端

            context.write(new Text(newWord), new IntWritable(1));

        }

    }

}

ScrewMapper.java 文件内容

 package cn.org.yinzhengjie.srew;

 import org.apache.hadoop.io.IntWritable;

 import org.apache.hadoop.io.LongWritable;

 import org.apache.hadoop.io.Text;

 import org.apache.hadoop.mapreduce.Mapper;

 import java.io.IOException;

 public class ScrewMapper2 extends Mapper<LongWritable,Text,Text,IntWritable> {

     //处理的数据类似于“1_1    677”

     @Override

     protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

         String line = value.toString();

         //

         String[] arr = line.split("\t");

         //newKey

         String newKey = arr[0].split("_")[0];

         //newVAl

         int newVal = Integer.parseInt(arr[1]);

         context.write(new Text(newKey), new IntWritable(newVal));

     }

 }

ScrewMapper2.java 文件内容

 /*

 @author :yinzhengjie

 Blog:http://www.cnblogs.com/yinzhengjie/tag/Hadoop%E8%BF%9B%E9%98%B6%E4%B9%8B%E8%B7%AF/

 EMAIL:y1053419035@qq.com

 */

 package cn.org.yinzhengjie.srew;

 import org.apache.hadoop.io.IntWritable;

 import org.apache.hadoop.io.Text;

 import org.apache.hadoop.mapreduce.Reducer;

 import java.io.IOException;

 public class ScrewReducer extends Reducer<Text,IntWritable,Text,IntWritable> {

     @Override

     protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {

         int count = 0;

         for (IntWritable value : values) {

             count += value.get();

         }

         context.write(key,new IntWritable(count));

     }

 }

ScrewReducer.java 文件内容

 /*

 @author :yinzhengjie

 Blog:http://www.cnblogs.com/yinzhengjie/tag/Hadoop%E8%BF%9B%E9%98%B6%E4%B9%8B%E8%B7%AF/

 EMAIL:y1053419035@qq.com

 */

 package cn.org.yinzhengjie.srew;

 import org.apache.hadoop.conf.Configuration;

 import org.apache.hadoop.fs.FileSystem;

 import org.apache.hadoop.fs.Path;

 import org.apache.hadoop.io.IntWritable;

 import org.apache.hadoop.io.Text;

 import org.apache.hadoop.mapreduce.Job;

 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

 public class ScrewApp {

     public static void main(String[] args) throws Exception {

         //实例化一个Configuration，它会自动去加载本地的core-site.xml配置文件的fs.defaultFS属性。(该文件放在项目的resources目录即可。)

         Configuration conf = new Configuration();

         //将hdfs写入的路径定义在本地，需要修改默认为文件系统，这样就可以覆盖到之前在core-site.xml配置文件读取到的数据。

         conf.set("fs.defaultFS","file:///");

         //代码的入口点，初始化HDFS文件系统，此时我们需要把读取到的fs.defaultFS属性传给fs对象。

         FileSystem fs = FileSystem.get(conf);

         //创建一个任务对象job，别忘记把conf穿进去哟！

         Job job = Job.getInstance(conf);

         //给任务起个名字

         job.setJobName("WordCount");

         //指定main函数所在的类，也就是当前所在的类名

         job.setJarByClass(ScrewApp.class);

         //指定map的类名,这里指定咱们自定义的map程序即可

         job.setMapperClass(ScrewMapper.class);

         //指定reduce的类名，这里指定咱们自定义的reduce程序即可

         job.setReducerClass(ScrewReducer.class);

         //设置输出key的数据类型

         job.setOutputKeyClass(Text.class);

         //设置输出value的数据类型

         job.setOutputValueClass(IntWritable.class);

         Path localPath = new Path("D:\\10.Java\\IDE\\yhinzhengjieData\\MyHadoop\\MapReduce\\out");

         if (fs.exists(localPath)){

             fs.delete(localPath,true);

         }

         //设置输入路径，需要传递两个参数，即任务对象(job)以及输入路径

         FileInputFormat.addInputPath(job,new Path("D:\\10.Java\\IDE\\yhinzhengjieData\\MyHadoop\\MapReduce\\screw.txt"));

         //设置输出路径，需要传递两个参数，即任务对象(job)以及输出路径

         FileOutputFormat.setOutputPath(job,localPath);

         //设置Reduce的个数为2.

         job.setNumReduceTasks(2);

         //等待任务执行结束,将里面的值设置为true。

         if (job.waitForCompletion(true)) {

             //当第一个MapReduce结束之后，我们这里又启动了一个新的MapReduce，逻辑和上面类似。

             Job job2 = Job.getInstance(conf);

             job2.setJobName("Wordcount2");

             job2.setJarByClass(ScrewApp.class);

             job2.setMapperClass(ScrewMapper2.class);

             job2.setReducerClass(ScrewReducer.class);

             job2.setOutputKeyClass(Text.class);

             job2.setOutputValueClass(IntWritable.class);

             Path p2 = new Path("D:\\10.Java\\IDE\\yhinzhengjieData\\MyHadoop\\MapReduce\\out2");

             if (fs.exists(p2)) {

                 fs.delete(p2, true);

             }

             FileInputFormat.addInputPath(job2, localPath);

             FileOutputFormat.setOutputPath(job2, p2);

             //我们将第一个MapReduce的2个reducer的处理结果放在新的一个MapReduce中只启用一个MapReduce。

             job2.setNumReduceTasks(1);

             job2.waitForCompletion(true);

         }

     }

 }

ScrewApp.java 文件内容

2>.检测实验结果

　　“D:\\10.Java\\IDE\\yhinzhengjieData\\MyHadoop\\MapReduce\\out” 目录内容如下：

Hadoop基础-MapReduce的数据倾斜解决方案

　　“D:\\10.Java\\IDE\\yhinzhengjieData\\MyHadoop\\MapReduce\\out2” 目录内容如下：

Hadoop基础-MapReduce的数据倾斜解决方案

四.解决数据倾斜方案之使用随机分区

1>.具体代码如下

 /*

 @author :yinzhengjie

 Blog:http://www.cnblogs.com/yinzhengjie/tag/Hadoop%E8%BF%9B%E9%98%B6%E4%B9%8B%E8%B7%AF/

 EMAIL:y1053419035@qq.com

 */

 package cn.org.yinzhengjie.screwpartition;

 import org.apache.hadoop.io.IntWritable;

 import org.apache.hadoop.io.LongWritable;

 import org.apache.hadoop.io.Text;

 import org.apache.hadoop.mapreduce.Mapper;

 import java.io.IOException;

 public class Screw2Mapper extends Mapper<LongWritable,Text,Text,IntWritable> {

     @Override

     protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

         String line = value.toString();

         String[] arr = line.split(" ");

         for(String word : arr){

             context.write(new Text(word), new IntWritable(1));

         }

     }

 }

Screw2Mapper.java 文件内容

 /*

 @author :yinzhengjie

 Blog:http://www.cnblogs.com/yinzhengjie/tag/Hadoop%E8%BF%9B%E9%98%B6%E4%B9%8B%E8%B7%AF/

 EMAIL:y1053419035@qq.com

 */

 package cn.org.yinzhengjie.screwpartition;

 import org.apache.hadoop.io.IntWritable;

 import org.apache.hadoop.io.Text;

 import org.apache.hadoop.mapreduce.Partitioner;

 import java.util.Random;

 public class Screw2Partition extends Partitioner<Text, IntWritable> {

     @Override

     public int getPartition(Text text, IntWritable intWritable, int numPartitions) {

         Random r = new Random();

         //返回的是分区的随机的一个ID

         return r.nextInt(numPartitions);

     }

 }

Screw2Partition.java 文件内容

 /*

 @author :yinzhengjie

 Blog:http://www.cnblogs.com/yinzhengjie/tag/Hadoop%E8%BF%9B%E9%98%B6%E4%B9%8B%E8%B7%AF/

 EMAIL:y1053419035@qq.com

 */

 package cn.org.yinzhengjie.screwpartition;

 import org.apache.hadoop.io.IntWritable;

 import org.apache.hadoop.io.Text;

 import org.apache.hadoop.mapreduce.Reducer;

 import java.io.IOException;

 public class Screw2Reducer extends Reducer<Text,IntWritable,Text,IntWritable> {

     @Override

     protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {

         int sum = 0;

         for(IntWritable value : values){

             sum += value.get();

         }

         context.write(key,new IntWritable(sum));

     }

 }

Screw2Reducer.java 文件内容

 /*

 @author :yinzhengjie

 Blog:http://www.cnblogs.com/yinzhengjie/tag/Hadoop%E8%BF%9B%E9%98%B6%E4%B9%8B%E8%B7%AF/

 EMAIL:y1053419035@qq.com

 */

 package cn.org.yinzhengjie.screwpartition;

 import org.apache.hadoop.conf.Configuration;

 import org.apache.hadoop.fs.FileSystem;

 import org.apache.hadoop.fs.Path;

 import org.apache.hadoop.io.IntWritable;

 import org.apache.hadoop.io.Text;

 import org.apache.hadoop.mapreduce.Job;

 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

 public class Screw2App {

     public static void main(String[] args) throws Exception {

         Configuration conf = new Configuration();

         conf.set("fs.defaultFS", "file:///");

         FileSystem fs = FileSystem.get(conf);

         Job job = Job.getInstance(conf);

         job.setJobName("Wordcount");

         job.setJarByClass(Screw2App.class);

         job.setMapperClass(Screw2Mapper.class);

         job.setReducerClass(Screw2Reducer.class);

         job.setPartitionerClass(Screw2Partition.class);

         job.setOutputKeyClass(Text.class);

         job.setOutputValueClass(IntWritable.class);

         Path p = new Path("D:\\10.Java\\IDE\\yhinzhengjieData\\MyHadoop\\MapReduce\\out");

         if (fs.exists(p)) {

             fs.delete(p, true);

         }

         FileInputFormat.addInputPath(job, new Path("D:\\10.Java\\IDE\\yhinzhengjieData\\MyHadoop\\MapReduce\\screw.txt"));

         FileOutputFormat.setOutputPath(job, p);

         job.setNumReduceTasks(2);

         job.waitForCompletion(true);

     }

 }

Screw2App.java 文件内容

2>.检测实验结果

　　“D:\\10.Java\\IDE\\yhinzhengjieData\\MyHadoop\\MapReduce\\out” 目录内容如下：

Hadoop基础-MapReduce的数据倾斜解决方案

　　 “D:\\10.Java\\IDE\\yhinzhengjieData\\MyHadoop\\MapReduce\\out2” 目录内容如下：

相关文章