MapReduce实验02——求平均值

MapReduce实验——求平均值

要求使用mapreduce统计出每类商品的平均点击次数

重点：输入文件的格式！

遇到的问题：

运行过程中报出java.lang.ArrayIndexOutOfBoundsException（数组越界）的错误

原因：

一行数据中间的空格格式不正确

/**
 * MapReduce实验——求平均值
 *     要求使用mapreduce统计出每类商品的平均点击次数
 * 
 * 重点：输入文件的格式！
 * 
 * 遇到的问题：
 *     运行过程中报出java.lang.ArrayIndexOutOfBoundsException（数组越界）的错误
 *     原因：一行数据中间的空格格式不正确
 */
package MapReducetests;

import java.io.IOException;

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.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;



public class MapReducetest2 {
    static String INPUT_PATH="hdfs://192.168.57.128:9000/testhdfs1026/run/input/mapreducetest2.txt";
    static String OUTPUT_PATH="hdfs://192.168.57.128:9000/testhdfs1026/run/output/test2";
    
    /*
     * Mapper
     * NullWritable是Writable的一个特殊类，序列化的长度为0，实现方法为空实现
     * 如果你不需要使用键或值，你就可以将键或值声明为NullWritable。
     */
    static class MyMapper extends Mapper<Object,Text,Text,IntWritable>
    {    
        //map将输入中的value复制到输出数据的key上，并直接输出
        private static Text newKey=new Text();
        protected void map(Object key, Text value, Context context) //Context context 记录输入的key和value 
                throws IOException, InterruptedException
        {   
            String line=value.toString();     // toString()方法返回反映这个对象的字符串
            System.out.println("Map.line:"+line);
            String arr[]=line.split(" ");    //split拆分
            newKey.set(arr[0]);
            int click = Integer.parseInt(arr[1]);
            context.write(newKey, new IntWritable(click));
        } 
    }   
    
    //Reducer
    static class MyReduce extends Reducer<Text,IntWritable,Text,IntWritable>
    {   
        protected void reduce(Text key,Iterable<IntWritable> values,Context context) 
                throws IOException,InterruptedException
        {
            int num = 0;
            int count = 0;
            //for(IntWritable val:values){} 对IntWritable    类型的集合进行遍历
            for(IntWritable val:values){
                num += val.get();
                count++;
            }
            int avg = num/count;
            context.write(key, new IntWritable(avg));
        }   
    }
    
    
    
    public static void main(String[] args) 
            throws IOException,ClassNotFoundException,InterruptedException{
        Path inputpath=new Path(INPUT_PATH);
        Path outputpath=new Path(OUTPUT_PATH);
        Configuration conf=new Configuration();
        System.out.println("Start");
        Job job=Job.getInstance(conf);
        job.setJarByClass(MapReducetest2.class);
        job.setMapperClass(MyMapper.class);
        job.setReducerClass(MyReduce.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        FileInputFormat.addInputPaths(job, INPUT_PATH);
        FileOutputFormat.setOutputPath(job,outputpath);
        
        boolean flag = job.waitForCompletion(true);
        /*
         * wait for completion 工作等待完成。
         * Job运行是通过job.waitForCompletion(true)，
         * true表示将运行进度等信息及时输出给用户，false的话只是等待作业结束
         */
        System.out.println(flag);
        System.exit(flag? 0 : 1);
    }
    
    
}