hadoop案例：wordcountcombine(wordcount合并）

我就是我 2022-11-21 15:17 176阅读 0赞

### 文章目录 ###

*   *  输入数据
     *  期望结果
     *  需求分析
     *  自定义WordcountCombiner
     *  Mapper类
     *  Reducer类
     *  Driver类
     *  执行结果

## 输入数据 ##

统计过程中对每一个MapTask的输出进行局部汇总，以减小网络传输量即采用Combiner功能  
hello.txt

banzhang ni hao
    xihuan hadoop banzhang
    banzhang ni hao
    xihuan hadoop banzhang

## 期望结果 ##

期望输出数据  
期望：Combine输入数据多，输出时经过合并，输出数据降低。

## 需求分析 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjc0MzYw_size_16_color_FFFFFF_t_70_pic_center]

## 自定义WordcountCombiner ##

package copyFiles.wordcount2;
    
    import java.io.IOException;
    import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Reducer;
    
    public class WordcountCombiner extends Reducer<Text, IntWritable, Text, IntWritable>{ 
    
    IntWritable v = new IntWritable();
    
    	@Override
    	protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { 
    
            // 1 汇总
    		int sum = 0;
    
    		for(IntWritable value :values){ 
    			sum += value.get();
    		}
    
    		v.set(sum);
    
    		// 2 写出
    		context.write(key, v);
    	}
    }

## Mapper类 ##

package copyFiles.wordcount2;
    
    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 WordCountMapper extends Mapper<LongWritable, Text,Text, IntWritable> { 
    
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { 
            //
            String string = value.toString();
            String[] split = string.split(" ");
            for (String s : split) { 
                new Text().set(s);
                context.write(new Text(),new IntWritable());
            }
    
    
        }
    }

## Reducer类 ##

package copyFiles.wordcount2;
    
    import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Reducer;
    
    import java.io.IOException;
    
    public class WordCountReduce 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();
    
            }
            new IntWritable().set(sum);
            //
            context.write(key,new IntWritable());
        }
    }

## Driver类 ##

package copyFiles.wordcount2;
    
    import com.mr.wordcount.WordCountMapper;
    import com.mr.wordcount.WordCountReducer;
    import org.apache.hadoop.conf.Configuration;
    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;
    
    import java.io.IOException;
    
    public class WordCountDriver  { 
        public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { 
    // 0 指定路径
            args = new String[]{ "E:\\Hadoop\\src\\main\\resources\\input\\hello", "E:\\Hadoop\\src\\main\\resources\\ouput\\hello"};
            //args = new String[]{"E:/Hadoop/src/main/resources/", "E:/Hadoop/src/main/resources/"};
    
    // 1 获取配置信息configuration以及封装任务job
            Configuration configuration = new Configuration();
    
            Job job = Job.getInstance(configuration);
    
    // 2 设置Driver加载路径 setJarByClass
            job.setJarByClass(WordCountDriver.class);
    // 3 设置map和reduce类 setMaper setReducer
            job.setMapperClass(WordCountMapper.class);
            job.setReducerClass(WordCountReducer.class);
    // 4 设置map输出 setmapoutputkey setmapoutputvalue
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(IntWritable.class);
    // 5 设置最终输出kv类型 (reducer的输出kv类型) setoutoutkey setoutputvalue
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(IntWritable.class);
            job.setCombinerClass(WordcountCombiner.class);
            job.setCombinerClass(WordCountReducer.class);
    // 6 设置本地的输入和输出路径 com.mr.fileinputformat.setinputpath
            FileInputFormat.setInputPaths(job, new Path(args[0]));
            FileOutputFormat.setOutputPath(job, new Path(args[1]));
    
            // 7 提交
            boolean completion = job.waitForCompletion(true);
            System.exit(completion ? 0 : 1);
        }
        }

## 执行结果 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjc0MzYw_size_16_color_FFFFFF_t_70_pic_center 1]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjc0MzYw_size_16_color_FFFFFF_t_70_pic_center 2]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjc0MzYw_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/7282e5f35df544fdaa835c376bb41c3c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjc0MzYw_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221120/c2e0431ea2a84e599b35f60e62c09124.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjc0MzYw_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221120/29003568397d48ebae8dc05c44398934.png