MapReduce instance (II): Average

Hello everyone , I am Fengyun , Welcome to my blog perhaps WeChat official account 【 Laugh at Fengyun Road 】, In the days to come, let's learn about big data related technologies , Work hard together , Meet a better self ！

Realize the idea

The average is MapReduce Common algorithms , The algorithm of finding the average is also relatively simple , One idea is Map End read data , Enter data into Reduce Go through before shuffle, take map Function output key All with the same value value Values form a set value-list, Then enter into Reduce End ,Reduce The end summarizes and counts the number of records , Then do business . The specific principle is shown in the figure below ：

Write code

Mapper Code

public static class Map extends Mapper<Object , Text , Text , IntWritable>{
      
  private static Text newKey=new Text();  
  // Realization map function  
  public void map(Object key,Text value,Context context) throws IOException, InterruptedException{
      
    //  Convert the data of the input plain text file into String 
    String line=value.toString();  
    System.out.println(line);  
    String arr[]=line.split("\t");  
    newKey.set(arr[0]);  
    int click=Integer.parseInt(arr[1]);  
    context.write(newKey, new IntWritable(click));  
  }  
}

map End in use Hadoop After the default input method , Will input value Value through split() Method to intercept , We convert the intercepted product Click times field into IntWritable Type and set it to value, Set the commodity classification field to key, And then directly output key/value Value .

Reducer Code

public static class Reduce extends Reducer<Text, IntWritable, Text, IntWritable>{
      
  // Realization reduce function  
  public void reduce(Text key,Iterable<IntWritable> values,Context context) throws IOException, InterruptedException{
      
    int num=0;  
    int count=0;  
    for(IntWritable val:values){
      
      num+=val.get(); // Sum each element num 
      count++;        // Count the number of elements count 
    }  
    int avg=num/count;  // Calculate the average  

    context.write(key,new IntWritable(avg));  
  }  
}  

map Output <key,value> after shuffle Process integration <key,values> Key value pair , And then <key,values> Key value pairs are given to reduce.reduce Termination received values after , Will input key Copy directly to the output key, take values adopt for Loop sums every element inside num And count the number of elements count, And then use num Divide count Get the average avg, take avg Set to value, Finally, output directly <key,value> That's all right. .

Complete code

package mapreduce;  
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.NullWritable;  
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.input.TextInputFormat;  
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;  
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;  
public class MyAverage{
      
  public static class Map extends Mapper<Object , Text , Text , IntWritable>{
      
    private static Text newKey=new Text();  
    public void map(Object key,Text value,Context context) throws IOException, InterruptedException{
      
      String line=value.toString();  
      System.out.println(line);  
      String arr[]=line.split("\t");  
      newKey.set(arr[0]);  
      int click=Integer.parseInt(arr[1]);  
      context.write(newKey, new IntWritable(click));  
    }  
  }  
  public static class Reduce extends Reducer<Text, IntWritable, Text, IntWritable>{
      
    public void reduce(Text key,Iterable<IntWritable> values,Context context) throws IOException, InterruptedException{
      
      int num=0;  
      int count=0;  
      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{
      
    Configuration conf=new Configuration();  
    System.out.println("start");  
    Job job =new Job(conf,"MyAverage");  
    job.setJarByClass(MyAverage.class);  
    job.setMapperClass(Map.class);  
    job.setReducerClass(Reduce.class);  
    job.setOutputKeyClass(Text.class);  
    job.setOutputValueClass(IntWritable.class);  
    job.setInputFormatClass(TextInputFormat.class);  
    job.setOutputFormatClass(TextOutputFormat.class);  
    Path in=new Path("hdfs://localhost:9000/mymapreduce4/in/goods_click");  
    Path out=new Path("hdfs://localhost:9000/mymapreduce4/out");  
    FileInputFormat.addInputPath(job,in);  
    FileOutputFormat.setOutputPath(job,out);  
    System.exit(job.waitForCompletion(true) ? 0 : 1);  

  }  
}

-------------- end ----------------

WeChat official account ： Below scan QR code or Search for Laugh at Fengyun Road Focus on