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