package cn.lmj.mapreduce;
import java.io.IOException;
import java.util.Iterator;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.TextOutputFormat;
public class WordCount
{
//mapper
public static class WordCountMapper extends MapReduceBase implements Mapper<LongWritable,Text,Text,LongWritable>
{
LongWritable count = new LongWritable(1);
Text content = new Text();
@Override
public void map(LongWritable key, Text value,
OutputCollector<Text, LongWritable> output, Reporter report)
throws IOException
{
//切割字符串
String str = value.toString();
String[] arr = str.split(" ");
for(String s : arr)
{
content.set(s);
output.collect(content,count);
}
}
}
//reducer
public static class WordCountReduce extends MapReduceBase implements Reducer<Text,LongWritable,Text,LongWritable>
{
@Override
public void reduce(Text key, Iterator<LongWritable> values,
OutputCollector<Text, LongWritable> output, Reporter rep)
throws IOException
{
//将同样key的value累加
long sum = 0;
while(values.hasNext())
{
sum+=values.next().get();
}
output.collect(key,new LongWritable(sum));
}
}
public static void main(String[] args) throws Exception
{
//创建一个JobConf
JobConf conf = new JobConf(WordCount2.class);
conf.setJobName("lmj");
//设置输出类型
conf.setOutputKeyClass(Text.class);
conf.setOutputValueClass(LongWritable.class);
//设置Map、Combine和Reduce处理类
conf.setMapperClass(WordCountMapper.class);
conf.setCombinerClass(WordCountReduce.class);
conf.setReducerClass(WordCountReduce.class);
//设置输入类型
conf.setInputFormat(TextInputFormat.class);
conf.setOutputFormat(TextOutputFormat.class);
//设置输入和输出文件夹
FileInputFormat.setInputPaths(conf,new Path("/aaa/hadoop.txt"));
FileOutputFormat.setOutputPath(conf,new Path("/aaa/output"));
//启动jobConf
JobClient.runJob(conf);
}
}