hadoop mapper reducer sample demo

package com.traveller.bumble.hadoop.mr.temperature;

import com.traveller.bumble.hadoop.mr.wordcount.WordCountMapper;
import com.traveller.bumble.hadoop.mr.wordcount.WordCountReducer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
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.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.File;
import java.io.IOException;

/**
* Created by macbook on 2017/5/23.
*/
public class TemperatureApp {

public static void main(String[] args) {

try {

String srcPath = args[0];
String destPath = args[1];
String isLocal = args[2];
boolean flag = false;
//本地模式
if (isLocal.equals("1")) {
File file = new File(destPath);
flag = deleteDir(file);
//集群模式
} else {

flag = deletePath(new Path(destPath));

}

System.out.println(flag);
Job job = Job.getInstance();
job.setJobName("temperature");
job.setJarByClass(TemperatureApp.class);

job.setNumReduceTasks(1);

FileInputFormat.addInputPath(job, new Path(srcPath));
FileOutputFormat.setOutputPath(job, new Path(destPath));
job.setMapperClass(TemperatureMapper.class);
job.setReducerClass(TemperatureReducer.class);

job.setOutputKeyClass(IntWritable.class);
job.setMapOutputValueClass(IntWritable.class);
job.waitForCompletion(true);

} catch (Exception e) {
e.printStackTrace();
}

}

private static boolean deletePath(Path path) {
try {
Configuration conf = new Configuration();
FileSystem fs = FileSystem.get(conf);
boolean flag = fs.deleteOnExit(path);
return flag;
} catch (IOException e) {
e.printStackTrace();
return false;
}
}

private static boolean deleteDir(File dir) {
if (dir.isDirectory()) {
String[] children = dir.list();
for (int i = 0; i < children.length; i++) {
boolean success = deleteDir(new File(dir, children[i]));
if (!success) {
return false;
}
}
}
return dir.delete();
}
}

package com.traveller.bumble.hadoop.mr.temperature;

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

import java.io.IOException;

/**
* Created by macbook on 2017/5/24.
*/
public class TemperatureMapper extends Mapper<LongWritable,Text,IntWritable,IntWritable> {

protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
String [] strs = value.toString().split(",");
IntWritable year = new IntWritable(Integer.valueOf(strs[0]));
IntWritable temprature = new IntWritable(Integer.valueOf(strs[1]));
context.write(year,temprature);

}
}

package com.traveller.bumble.hadoop.mr.temperature;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;

/**
* Created by macbook on 2017/5/24.
*/
public class TemperatureReducer extends Reducer<IntWritable,IntWritable,IntWritable,IntWritable> {

protected void reduce(IntWritable key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {

Iterator<IntWritable> iterators = values.iterator();
List<Integer> temp = new ArrayList<>();
while (iterators.hasNext()){
temp.add(iterators.next().get());
}
Integer max = Collections.max(temp);
context.write(key,new IntWritable(max));
}
}