hadoop的map是具有输出自动排序功能的~继续学习~
import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configured; 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.input.TextInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.util.Tool; import org.apache.hadoop.util.ToolRunner; import java.io.IOException; public class Sort extends Configured implements Tool { //这里map将输入的value转化成IntWritable类型,作为输出的key public static class Map extends Mapper<Object,Text,IntWritable,IntWritable> { private static IntWritable data = new IntWritable(); public void map(Object key,Text value,Context context) throws IOException,InterruptedException { String line = value.toString(); System.out.println("line" + line); data.set(Integer.parseInt(line)); context.write(data, new IntWritable(1)); } } //reduce将输入的key复制到输出的value上,然后根据输入的value-list中的元素的个数决定key的输出次数 public static class Reduce extends Reducer<IntWritable,IntWritable,IntWritable,IntWritable> { //全局linenum来代表key的位次 private static IntWritable linenum = new IntWritable(1); public void reduce(IntWritable key,Iterable<IntWritable> values,Context context) throws IOException,InterruptedException{ for(IntWritable val : values){ context.write(linenum,key); System.out.println(linenum+" "+key); linenum = new IntWritable(linenum.get()+1); } } } public int run(String[] args) throws Exception{ Configuration aaa = new Configuration(); Job job = Job.getInstance(aaa); String InputPaths = "/usr/local/idea-IC-139.1117.1/Hadoop/out/datainput/sort.txt"; String OutputPath = "/usr/local/idea-IC-139.1117.1/Hadoop/out/dataout/"; job.setJarByClass(Sort.class); job.setJobName("Sort"); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); job.setMapperClass(Map.class); job.setReducerClass(Reduce.class); FileInputFormat.setInputPaths(job, new Path(InputPaths)); FileOutputFormat.setOutputPath(job, new Path(OutputPath)); job.setOutputKeyClass(IntWritable.class); job.setOutputValueClass(IntWritable.class); boolean success = job.waitForCompletion(true); return success ? 0 : 1; } public static void main(String[] args) throws Exception{ int ret = ToolRunner.run(new Sort(),args); System.exit(ret); } }
时间: 2024-10-14 18:39:06