packagecom.mr.ch07.maxmin;
importjava.io.DataInput;
importjava.io.DataOutput;
importjava.io.IOException;
importorg.apache.hadoop.conf.Configuration;
importorg.apache.hadoop.fs.Path;
importorg.apache.hadoop.io.Text;
importorg.apache.hadoop.io.Writable;
importorg.apache.hadoop.mapreduce.Job;
importorg.apache.hadoop.mapreduce.Mapper;
importorg.apache.hadoop.mapreduce.Reducer;
importorg.apache.hadoop.mapreduce.lib.input.FileInputFormat;
importorg.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
publicclassMinMaxValueDemo7_2_1 {
publicstaticclassMinMaxMapperextendsMapper<Object, Text, Text, MinMaxWritable> {
privateMinMaxWritableoutTuple=newMinMaxWritable();
@Overridepublicvoidmap(Objectkey, Textvalue, Contextcontext) throwsIOException, InterruptedException {
String[] strs=value.toString().split(" ");
StringstrDate=strs[0];// 定义记录日期的字符串变量strDateif (strDate==null) {
return;// 如果该日期值为空，则返回    }
System.out.println(strs[0] +",,,,"+strs[1]);
// 将值即做为最大值又做为最小值存储到自定义Writable类MinMaxWritable中。outTuple.setMin(Integer.parseInt(strs[1]));
outTuple.setMax(Integer.parseInt(strs[1]));
// 将结果写入上下文context.write(newText(strDate), outTuple);
  }
    }
publicstaticclassMinMaxReducerextendsReducer<Text, MinMaxWritable, Text, MinMaxWritable> {
privateMinMaxWritableresult=newMinMaxWritable();
@Overridepublicvoidreduce(Textkey, Iterable<MinMaxWritable>values, Contextcontext) 
throwsIOException, InterruptedException {
result.setMax(0);
result.setMin(0);
// 按key迭代输出value的值for (MinMaxWritableval : values) {
// 最小值放于结果集中if (result.getMin() ==0||val.getMin() <result.getMin()) {
result.setMin(val.getMin());
    }
// 最大值放于结果集中if (result.getMax() ==0||val.getMax() >result.getMax()) {
System.out.println("val.getMax(): "+val.getMax() +",,,,"+result.getMax());
result.setMax(val.getMax());
    }
    }
context.write(key, result);
  }
    }
publicstaticvoidmain(String[] args) throwsException {
Configurationconf=newConfiguration();
// String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();String[] otherArgs=newString[] { "/input/ch07/minmax", "/output/ch07/minmax" };
if (otherArgs.length!=2) {
System.err.println("Usage: MinMaxCountDriver <in> <out>");
System.exit(2);
  }
// Job job = new Job(conf, "StackOverflow Comment Date Min Max Count");Jobjob=Job.getInstance(conf);
job.setJarByClass(MinMaxValueDemo7_2_1.class);
job.setMapperClass(MinMaxMapper.class);
job.setCombinerClass(MinMaxReducer.class);
job.setReducerClass(MinMaxReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(MinMaxWritable.class);
FileInputFormat.addInputPath(job, newPath(otherArgs[0]));
FileOutputFormat.setOutputPath(job, newPath(otherArgs[1]));
System.exit(job.waitForCompletion(true) ?0 : 1);
    }
publicstaticclassMinMaxWritableimplementsWritable {
privateintmin;// 记录最小值privateintmax;// 记录最大值publicintgetMin() {
returnmin;
  }
publicvoidsetMin(intmin) {
this.min=min;
  }
publicintgetMax() {
returnmax;
  }
publicvoidsetMax(intmax) {
this.max=max;
  }
@OverridepublicvoidreadFields(DataInputin) throwsIOException {
min=in.readInt();
max=in.readInt();
  }
@Overridepublicvoidwrite(DataOutputout) throwsIOException {
out.writeInt(max);
out.writeInt(min);
    }
@OverridepublicStringtoString() {
returnmax+"\t"+min;
  }
    }
}