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);

}

}