MapReduce代码
Map过程
IntWritable one = new IntWritable(1);
public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException{
String []res=value.toString().split("\t");
String url=res[28];
context.write(new Text(url), one);
}
Reduce过程
public void reduce(Text key, Iterable values,
Context context) throws IOException, InterruptedException{
int sum = 0;
for(IntWritable value:values){
sum+=value.get();
}
context.write(new Text(key.toString()+":"+sum),NullWritable.get());
}
job过程
public staticvoid main(String[] args) throws Exception{
Configurationconf =new Configuration();
Jobjob= Job.getInstance(conf,"Pv");
job.setJarByClass(PvUvjob.class);
Pathin=new Path("/user/input/pv");
Pathout=new Path("/user/output/pv");
FileInputFormat.addInputPath(job,in);
FileOutputFormat.setOutputPath(job, out);
job.setInputFormatClass(TextInputFormat.class);
job.setOutputFormatClass(TextOutputFormat.class);
job.setMapperClass(PvUvMap.class);
job.setReducerClass(PvUvReduce.class);
//map输出类型
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(IntWritable.class);
//reduce输出类型
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(NullWritable.class);
job.setNumReduceTasks(1);
job.waitForCompletion(true);
}
?3�U�ta
