hadoop入门之利用hadoop来对文档数据归类统计案例wordcount

package pack;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
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;
public class sortCount {
public static void main(String [] args) throws IOException
{
Configuration conf =new Configuration();
Job job=new Job(conf);
job.setJarByClass(sortCount.class);
FileInputFormat.addInputPath(job, new Path("/anaconda.syslog"));
//必须有，这就是示例文件
FileOutputFormat.setOutputPath(job,new Path("/output") );
//必须没有，输出的结果
job.setMapperClass(myMapper.class);
//设置Map类
job.setReducerClass(myReducer.class);
//设置Reducer类
job.setCombinerClass(myReducer.class);
//预先为reducer处理数据，见下文详解
job.setNumReduceTasks(2);
//设置两个Reduce来执行任务
job.setPartitionerClass(myPartitioner.class);
job.setOutputKeyClass(Text.class);设置输出的键的类型
job.setOutputValueClass(IntWritable.class);设置输出的值的类型
try {
System.exit(job.waitForCompletion(true) ? 0 : 1);  //执行语句
} catch (ClassNotFoundException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}

package pack;
import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
public class myMapper extends Mapper<LongWritable, Text, Text, IntWritable>{
protected void map(LongWritable key ,Text value ,Context context) throws IOException, InterruptedException
{
//Text value 是一整行的数据     MAp源码会自动循环调用此函数只到把每一行都读入进来。
String [] str =value.toString().split(" ");
//for(String ss:str)
//{
//context.write(new Text(ss),new IntWritable(1));   //不知道把结果发给谁，用上下文
//}
//}//此注释代码用于wordCount示例
Text ss = new Text();
ss.set(new StringBuffer("loglevel::").append(str[1]).toString());//添加一个头部，用于区分
Text ss2 = new Text();
ss2.set(new StringBuffer("logresource::").append(str[2]).toString());
context.write(ss,new IntWritable(1));
context.write(ss2,new IntWritable(1));
//统计anaconda.syslog文件的各种提示出现次数
}}

package pack;
import java.io.IOException;
import java.util.Iterator;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Mapper.Context;
public class myReducer extends Reducer<Text, IntWritable, Text, IntWritable>{
//到reduce之前combiner已经自动把相同值（key）的数放到一起，现在只需要将其个数相加就可以了。
protected void reduce(Text key,Iterable<IntWritable> values, Context context) throws IOException, InterruptedException
{
int sum=0;
Iterator<IntWritable> it = values.iterator();
while(it.hasNext())
{
sum+=it.next().get();
}
context.write(key, new IntWritable(sum));
}
}
//ctrl+shift+t 查找类——>小技巧

package pack;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;
public class myPartitioner extends Partitioner<Text, IntWritable>{
@Override
public int getPartition(Text key, IntWritable value, int arg2) //arg2为设置的reducer数量{
// TODO Auto-generated method stub
if(key.toString().startsWith("loglevel::"))
return 0;//分发到第一个reducer
else if(key.toString().startsWith("logresource::"))
return 1;//分发到第二个reducer    0，1要以次写出来
return 0;
}
}