map reduce程序示例

 

package test2;import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.fs.Path;import org.apache.hadoop.io.IntWritable;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 java.io.IOException;/** 样例数据中包含了年份和温度，提出年份里温度最大的 (0, 0067011990999991950051507+0000+), (33, 0043011990999991950051512+0022+), (66, 0043011990999991950051518-0011+), (99, 0043012650999991949032412+0111+), (132, 0043012650999991949032418+0078+), (165, 0067011990999991937051507+0001+), (198, 0043011990999991937051512-0002+), (231, 0043011990999991945051518+0001+), (264, 0043012650999991945032412+0002+), (297, 0043012650999991945032418+0078+), * */public class mytest {static String INPUT_PATH="input/t1_num.txt";   //待统计的文件路径static String OUTPUT_PATH="output/t1_num";    //统计结果存放的路径static class MyMapper extends Mapper 
     
       {     //定义继承mapper类    protected void map(Object key, Object value, Context context) throws IOException, InterruptedException{    //定义map方法    String[] arr=value.toString().split("\\),");      //文件中的单词是以“),”分割的，并将每一行定义为一个数组    for(int i=0;i
      
       {     //定义继承reducer类    protected void reduce(Text key,Iterable
       
         values,Context context) throws IOException,InterruptedException{      //定义reduce方法         int max = 0;         for(IntWritable c:values){     //统计同一个单词的数量             if(c.get()>max){                 max = c.get();//获取value值             }         }        IntWritable outValue=new IntWritable(max);//挨个输出        context.write(key,outValue);     }    /**     在reduce过程，将map过程中的输出，按照相同的key(年份)将value放到同一个列表中作为reduce的输入     (1950, [0, 22, –11])     (1949, [111, 78])     (1937, [1, -2])     (1945, [1, 2, 78])     在reduce过程中，在列表中选择出最大的温度，将年-max温度的key-value作为输出：     (1950, 22)     (1949, 111)     (1937, 1)     (1945, 78)     */} public static void main(String[] args) throws Exception{    //main函数     System.setProperty("hadoop.home.dir", "D:\\hadoop-2.7.6");//这一行一定要     Path outputpath=new Path(OUTPUT_PATH);    //输出路径     Configuration conf=new Configuration();     Job job=Job.getInstance(conf);     //定义一个job，启动任务     FileInputFormat.setInputPaths(job, INPUT_PATH);     FileOutputFormat.setOutputPath(job,outputpath);     job.setMapperClass(MyMapper.class);     job.setReducerClass(MyReduce.class);     job.setOutputKeyClass(Text.class);     job.setOutputValueClass(IntWritable.class);     job.waitForCompletion(true);    }}