Hadoop-1.2.1 QQ推荐好友例子

package com.bjsxt.qq;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

/*
*
* QQ推荐好友
* 处理方案：在mapper中将key和value分别对调输出;在reduce中将mapper输出的集合进行笛卡尔输出
*/
public class QqMapper extends Mapper<LongWritable, Text, Text, Text>{
protected void map(LongWritable key, Text value,Context context) throws java.io.IOException ,InterruptedException {
String line =value.toString();
String[] ss=line.split(" ");
context.write(new Text(ss[0]), new Text(ss[1]));
context.write(new Text(ss[1]), new Text(ss[0]));
};
}






package com.bjsxt.qq;

import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

/*
*
* QQ推荐好友
*/
public class QqReduce extends Reducer<Text, Text, Text, Text>{
protected void reduce(Text key, java.lang.Iterable<Text> iterable,Context context) throws java.io.IOException ,InterruptedException {
Set<String> set=new HashSet<String>();
for (Text t: iterable) {
set.add(t.toString());
}

if(set.size() >1){
for (Iterator j=set.iterator();j.hasNext();) {
String name=(String)j.next();
for (Iterator k=set.iterator();k.hasNext();) {
String other=(String)k.next();
if(!name.equals(other)){
context.write(new Text(name), new Text(other));
}
}

}
}
};
}







package com.bjsxt.qq;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
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 QqJobRun {
public static void main(String[] args) {
Configuration conf=new Configuration();
conf.set("fs.default.name", "hdfs://192.168.0.200:9000");
conf.set("mapred.job.tracker", "192.168.0.200:9001");

try {
Job job=new Job(conf,"QQ-Job1");
job.setJarByClass(QqJobRun.class);
job.setMapperClass(QqMapper.class);
job.setReducerClass(QqReduce.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);

FileInputFormat.addInputPath(job, new Path("/opt/input/qq/"));
FileOutputFormat.setOutputPath(job, new Path("/opt/output/qq/"));

System.exit(job.waitForCompletion(true)?0:1);


} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}


}
}