MapReduce实验——计算整数的最大值最小值

计算整数的最大值最小值

Map类

package MaxMin_03;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

public class MyMap extends
        Mapper<LongWritable,Text, Text,LongWritable> {
    //为map设置一个固定的输出键
    private final Text keyText = new Text("K");
    private LongWritable val = new LongWritable(0);
    @Override
    protected void map(LongWritable key,Text value,Context context) throws IOException, InterruptedException {
        //为val对象设置读取的数值
        val.set(Long.parseLong(value.toString()));
        context.write(keyText,val);
    }
}

Reduce类

package MaxMin_03;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

public class MyReduce extends Reducer<Text, LongWritable,Text,Text> {
    //设置变量max和min的初始值
    private long max = Long.MIN_VALUE;
    private long min = Long.MAX_VALUE;
    @Override
    protected void reduce(Text key,Iterable<LongWritable> valueIn,Context context) throws IOException, InterruptedException {
        //循环得出最大值和最小值
        for(LongWritable val : valueIn){
            long tmp = val.get();
            if(tmp > max){
                max = tmp;
            }
            if(tmp < min){
                min = tmp;
            }
        }
        //最大值和最小值分别输出
        context.write(new Text("Max"),new Text(max + ""));
        context.write(new Text("Min"),new Text(min + ""));

    }
}

Job类

package MaxMin_03;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import org.apache.hadoop.io.Text;
import java.io.IOException;

public class TestJob {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration conf = new Configuration();
        //1、获取作业对象
        Job job = Job.getInstance(conf);
        //2、设置主类
        job.setJarByClass(TestJob.class);
        //3、设置job参数
        job.setMapperClass(MyMap.class);
        job.setReducerClass(MyReduce.class);
        //4 set map reduce output type
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(LongWritable.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(LongWritable.class);
        //5、设置job输入输出
        FileInputFormat.setInputPaths(job,new Path("file:///simple/maxmin.txt"));
        FileOutputFormat.setOutputPath(job,new Path("file:///simple/result"));
        //6 commit job
        System.out.println(job.waitForCompletion(true) ? 0 : 1);;
    }
}