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

gaoke

　　本博客属原创文章,转载请注明出处:http://guoyunsky.iyunv.com/blog/1235936
　　
　　请先阅读:           
　　1.Hadoop MapReduce 学习笔记(一) 序言和准备
　　2.Hadoop MapReduce 学习笔记(二) 序言和准备 2
　　3.Hadoop MapReduce 学习笔记(三) MapReduce实现类似SQL的SELECT MAX(ID)
　　4.Hadoop MapReduce 学习笔记(四) MapReduce实现类似SQL的SELECT MAX(ID) 2 一些改进
　　5.Hadoop MapReduce 学习笔记(五) MapReduce实现类似SQL的max和min
　　下一篇:
　　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();
}

}
}

 
更多文章、感悟、分享、勾搭，请用微信扫描: