Hadoop MapReduce 学习笔记(六) MapReduce实现类似SQL的max和min 正确写法

      以上写法得到的结果是错误的,因为基本延续了输出单个值的写法.而hadoop要输出多个值写法是不一样的.具体还是查看代码吧.

package com.guoyun.hadoop.mapreduce.study;

import java.io.IOException;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
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;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 或得最大和最小值，类似SQL：SELECT MAX(NUMBER),MIN(NUMBER) FROM TABLE
 * 注意：这里只有一列数据，多列请查看 @GetMaxAndMinValueMultiMapReduceTest
 */
public class GetMaxAndMinValueMapReduceFixTest extends MyMapReduceSIngleColumnTest{
  
  public static final Logger log=LoggerFactory.getLogger(GetMaxAndMinValueMapReduceFixTest.class);
  

  public GetMaxAndMinValueMapReduceFixTest(String outputPath) {
    super(outputPath);
    // TODO Auto-generated constructor stub
  }

  /**
   * Map,to get the source datas
   */
  public static class MyMapper extends Mapper<LongWritable,Text,Text,LongWritable>{
    private final Text writeKey=new Text("K");
    private LongWritable writeValue=new LongWritable(0);
    
    @Override
    protected void map(LongWritable key, Text value, Context context)
        throws IOException, InterruptedException {
      log.debug("begin to map");
      StringTokenizer tokenizer=null;
      String lineValue=null;
      
      
      tokenizer=new StringTokenizer(value.toString().trim());
      while(tokenizer.hasMoreTokens()){
        lineValue=tokenizer.nextToken().trim();
        if(lineValue.equals("")){
          continue;
        }
        try {
          writeValue.set(Long.parseLong(lineValue));
          context.write(writeKey, writeValue);
        } catch (NumberFormatException e) {
          continue;
        }
        
      }
    }
  }
  
  public static class MyCombiner
    extends Reducer<Text,LongWritable,Text,LongWritable>{
    private final Text maxValueKey=new Text("maxValue");
    private final Text minValueKey=new Text("minValue");
    
    @Override
    public void reduce(Text key, Iterable<LongWritable> values,Context context)
        throws IOException, InterruptedException {
      log.debug("begin to combine");
      long maxValue=Long.MIN_VALUE;
      long minValue=Long.MAX_VALUE;
      long valueTmp=0;
      LongWritable writeValue=new LongWritable(0); 
      for(LongWritable value:values){
        valueTmp=value.get();
        if(valueTmp>maxValue){
          maxValue=valueTmp;
        }else if(valueTmp<minValue){
          minValue=valueTmp;
        }
      }
      writeValue.set(maxValue);
      context.write(maxValueKey, writeValue);
      writeValue.set(minValue);
      context.write(minValueKey, writeValue);
    } 
    
  }
  
  
  /**
   * Reduce,to get the max value
   */
  public static class MyReducer 
    extends Reducer<Text,LongWritable,Text,LongWritable>{
    private final Text maxValueKey=new Text("maxValue");
    private final Text minValueKey=new Text("minValue");
      
    @Override
    public void run(Context context) throws IOException, InterruptedException {
      long maxValue=Long.MIN_VALUE;
      long minValue=Long.MAX_VALUE;
      long tmpValue=0;
      String tmpKey="";
      LongWritable writeValue=new LongWritable(0); 
      
      try {
        setup(context);
   
        while(context.nextKey()){
          tmpKey=context.getCurrentKey().toString();
          tmpValue=context.getCurrentValue().get();
          if(tmpKey.equals("maxValue")){
            if(tmpValue>maxValue){
              maxValue=tmpValue;
            }
          }else if(tmpKey.equals("minValue")){
            if(tmpValue<minValue){
              minValue=tmpValue;
            }
          }
        }
        
        writeValue.set(maxValue);
        context.write(maxValueKey, writeValue);
        writeValue.set(minValue);
        context.write(minValueKey, writeValue);
      } catch (Exception e) {
        log.debug(e.getMessage());
      }finally{
        cleanup(context);
      }
       
    }

    @Override
    protected void cleanup(Context context) throws IOException,
        InterruptedException {
      // TODO Auto-generated method stub
      super.cleanup(context);
    }

    @Override
    protected void setup(Context context) throws IOException,
        InterruptedException {
      // TODO Auto-generated method stub
      super.setup(context);
    }

   /* @Override
    public void reduce(Text key, Iterable<LongWritable> values,Context context)
        throws IOException, InterruptedException {
      log.debug("begin to reduce");
      long maxValue=Long.MIN_VALUE;
      long minValue=Long.MAX_VALUE;
      long valueTmp=0;
      LongWritable writeValue=new LongWritable(0); 
      for(LongWritable value:values){
        valueTmp=value.get();
        if(valueTmp>maxValue){
          maxValue=valueTmp;
        }else if(valueTmp<minValue){
          minValue=valueTmp;
        }
      }
      writeValue.set(maxValue);
      context.write(maxValueKey, writeValue);
      writeValue.set(minValue);
      context.write(minValueKey, writeValue);
    } */
    
  }
  
  /**
   * @param args
   */
  public static void main(String[] args) {
    MyMapReduceTest mapReduceTest=null;
    Configuration conf=null;
    Job job=null;
    FileSystem fs=null;
    Path inputPath=null;
    Path outputPath=null;
    long begin=0;
    String output="testDatas/mapreduce/MROutput_SingleColumn_getMaxAndMinFix";
    
    
    try {
      mapReduceTest=new GetMaxAndMinValueMapReduceTest(output);
      
      inputPath=new Path(mapReduceTest.getInputPath());
      outputPath=new Path(mapReduceTest.getOutputPath());
      
      conf=new Configuration();
      job=new Job(conf,"getMaxAndMinValueFix");
      
      fs=FileSystem.getLocal(conf);
      if(fs.exists(outputPath)){
        if(!fs.delete(outputPath,true)){
          System.err.println("Delete output file:"+mapReduceTest.getOutputPath()+" failed!");
          return;
        }
      }
      
      
      job.setJarByClass(GetMaxAndMinValueMapReduceFixTest.class);
      job.setMapOutputKeyClass(Text.class);
      job.setMapOutputValueClass(LongWritable.class);
      job.setOutputKeyClass(Text.class);
      job.setOutputValueClass(LongWritable.class);
      job.setMapperClass(MyMapper.class);
      job.setCombinerClass(MyCombiner.class);
      job.setReducerClass(MyReducer.class);
      
      job.setNumReduceTasks(2);
      
      FileInputFormat.addInputPath(job, inputPath);
      FileOutputFormat.setOutputPath(job, outputPath);
      
      begin=System.currentTimeMillis();
      job.waitForCompletion(true);
      
      System.out.println("=================================================================================");
      if(mapReduceTest.isGenerateDatas()){
        System.out.println("The maxValue is:"+mapReduceTest.getMaxValue());
        System.out.println("The minValue is:"+mapReduceTest.getMinValue());
      }
      System.out.println("Spend time:"+(System.currentTimeMillis()-begin));
      // Spend time:13361
      
    } catch (Exception e) {
      // TODO Auto-generated catch block
      e.printStackTrace();
    }
    

  }
}