大数据-Hadoop-MapReduce(二)：MapReduce编程案例

梦里梦外; 2023-10-05 17:39 15阅读 0赞

![在这里插入图片描述][20210125191658808.png]

## 案例：使用MapReduce进行词频统计 ##

### 1、读取本地数据，使用本地(Windows中的hadoop)计算资源，计算结果保存到本地 ###

![在这里插入图片描述][20210130135908461.png]

WCMapper.java

package com.wyr.wordcount;
    
    import java.io.IOException;
    import java.util.List;
    
    import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.InputSplit;
    import org.apache.hadoop.mapreduce.Mapper;
    import org.apache.hadoop.mapreduce.RecordReader;
    
    /**
     * 注意：导包时，导入 org.apache.hadoop.mapreduce包下的类(2.0的新api)
     * 
     * 1. 自定义的类必须符合 MapperReduce 的Mapper的规范
     * 
     * 2.在MapperReduce中，只能处理 key-value格式的数据
     * 	 KEYIN, VALUEIN： mapper输入的k-v类型。 由当前Job的 InputFormat 的 RecordReader决定！封装输入的 key-value 由 RecordReader 自动进行。
     *   KEYOUT, VALUEOUT： mapper输出的k-v类型: 自定义
     *   
     * 3. InputFormat的作用：
     *  		①验证输入目录中文件格式，是否符合当前Job的要求
     *  		②生成切片，每个切片都会交给一个MapTask处理；方法： List<InputSplit> getSplits
     *  		③创建RecordReader，由RecordReader从切片中读取记录，交给Mapper进行处理；方法：RecordReader<K,V> createRecordReader；默认hadoop使用的是TextInputFormat，TextInputFormat使用LineRecordReader!
     *
     * 4. 在Hadoop中，如果有Reduce阶段。通常key-value都需要实现序列化协议，来进行不同机器间的数据网络传输。
     *  	MapTask处理后的key-value，只是一个阶段性的结果！这些key-value需要传输到ReduceTask所在的机器！
     *  	不同机器间的数据传输最快捷的方式：将一个对象通过序列化技术，序列化到一个文件中，经过网络传输到另外一台机器，再使用反序列化技术，从文件中读取数据，还原为对象！
     *  	java的序列化协议的缺点： Serilizxxxxx，特点：不仅保存对象的属性值，类型，还会保存大量的包的结构，子父类和接口的继承信息！	保存的信息太多、太重
     *  	hadoop开发了一款轻量级的序列化协议： Wriable机制！
     *
     */
    public class WCMapper extends Mapper<LongWritable, Text, Text, IntWritable>{
       
        	// KEYIN, VALUEIN, KEYOUT, VALUEOUT    第 3 个参数表示单词；第 4个参数表示词频
    	private Text out_key=new Text();
    	private IntWritable out_value=new IntWritable(1);
    	// 针对输入的每个 keyin-valuein调用一次   （0，hello	hi	hello	hi），其中key为：0，value为：hello	hi	hello	hi
    	@Override
    	protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
       
        	// key 为 输入数据的每行的偏移量；value 为输入数据的每行的数据；context为输出数据
    		System.out.println("keyin:"+key+"----keyout:"+value);
    		String[] words = value.toString().split("\t");
    		for (String word : words) {
       
        
    			out_key.set(word);
    			//写出数据(单词,1)
    			context.write(out_key, out_value);
    		}	
    	}
    }

WCReducer.java

package com.wyr.wordcount;
    
    import java.io.IOException;
    
    import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Reducer;
    
    /**
     * 1. Reducer需要符合Hadoop的Reducer规范
     *
     * KEYIN, VALUEIN: Mapper输出的 keyout-valueout
     * KEYOUT, VALUEOUT: 自定义
     *
     */
    public class WCReducer extends Reducer<Text, IntWritable, Text, IntWritable>{
       
           // KEYIN, VALUEIN, KEYOUT, VALUEOUT
        private IntWritable out_value=new IntWritable();
        // reduce一次处理一组数据，key相同的视为一组
        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {
       
        
            int sum=0;
            for (IntWritable intWritable : values) {
       
        
                sum+=intWritable.get();
            }
            out_value.set(sum);
            //将累加的值写出
            context.write(key, out_value);
        }
    }

WCDriver.java

package com.wyr.wordcount;
    
    import java.io.IOException;
    import java.net.URI;
    
    import org.apache.hadoop.conf<

[20210125191658808.png]: https://img-blog.csdnimg.cn/20210125191658808.png
[20210130135908461.png]: https://img-blog.csdnimg.cn/20210130135908461.png