利用hadoop计算文件中的最大值

本文主要是找出文件中的最大值假设文件的类型为：

A 66

B 72

C 91

A 43

C 62

C 85

B 35

A 23

B 39

找出文件中的每个对象的最大值:

A 66

B 72

C 91

操作代码解析如下：

import java.io.IOException;
import java.net.URI;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
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.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class MaxNumMR extends Configured implements Tool{

@Override
public int run(String[] args) throws Exception {
//获取文件系统的配置文件
Configuration conf = new Configuration() ;
URI uri = new URI("hdfs://hadoop:9000") ;
//声明文件系统的对象
FileSystem fs = FileSystem.get(uri, conf) ;
//结果的输出目录，如果该目录存在则删除
Path path = new Path("hdfs://hadoop:9000/ming/out2") ;
if(fs.isDirectory(path)){
fs.delete(path, true) ;
}
//声明作业对象
Job job = new Job(conf);
job.setJarByClass(MaxNumMR.class);

//配置Mapper的类和输出的格式
job.setMapperClass(MaxMapper.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(Text.class);

//配置Reducer的类和输出的格式
job.setReducerClass(MaxReducer.class) ;
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);

//配置文件的输入和输出文件
FileInputFormat.addInputPath(job, new Path("hdfs://hadoop:9000/ming/math.txt"));
FileOutputFormat.setOutputPath(job, path);

return job.waitForCompletion(true)?0:1;
}

public static void main(String[] args) throws Exception {
//主方法，代码的入口点
int e = ToolRunner.run(new Configuration(), new MaxNumMR(), args);
System.exit(e);
}

public static class MaxMapper extends Mapper<LongWritable,Text,Text,Text>{
@Override
protected void map(LongWritable key,Text value,Context context) throws IOException, InterruptedException{
String line = value.toString() ;
//对数据进行切分，然后存入输出的文件中
String[] param = line.split("\t") ;
context.write(new Text(param[0]),new Text(param[1])) ;
}
}

public static class MaxReducer extends Reducer<Text,Text,Text,Text>{
@Override
protected void reduce(Text key,Iterable<Text> values,Context context) throws IOException, InterruptedException{
long max = 0L ;
for(Text t:values){
//判断数据的大小
if(Long.parseLong(t.toString()) > max){
max = Long.parseLong(t.toString()) ;
}
}
context.write(key, new Text(String.valueOf(max)));
}
}
}