如何通过MapReduce实现腾讯QQ推荐好友

导读：
如果我们会了基本的mapreduce模型，不论是qq好友，还是文章推荐，道理都是一样的。只要我们勤于思考，能创造出更多例子。

大家都知道qq用户量上亿，每个用户又有很多的好友，因此，数据量十分的庞大，如何才能实现QQ的好友推荐呢？
下面举一个例子：
A有QQ好友B
B有QQ好友C
则A，C有可能是好友。
当A登录的时候，则会向A推荐C，当C登录的时候，则会向C推荐A。

Demo

输入数据

[mw_shl_code=bash,true]tom jason
james ivy
kette jason
tom lgd
lgd lkx
lgd lyn[/mw_shl_code]

[mw_shl_code=bash,true]map阶段
key：主
value：从
key：从
value：主
将一条记录分别作为key，value进行输出。
tom-->jason
jason-->tom
tom-->lgd
lgd-->tom

reduce阶段
将同一个key的values值进行两两组合。[/mw_shl_code]

[mw_shl_code=java,true]package FriendsRecommended;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class FindFriends {
private final static String INPUT_PATH = "hdfs://liguodong:8020/liguodong";
private final static String OUTPUT_PATH = "hdfs://liguodong:8020/liguodong/QQFriendRecommended";

public static void main(String[] args) throws IOException,
URISyntaxException, ClassNotFoundException, InterruptedException {

      Configuration conf = new Configuration();
      final FileSystem fileSystem = FileSystem.get(new URI(INPUT_PATH),conf);
      if(fileSystem.exists(new Path(OUTPUT_PATH)))
      {
         fileSystem.delete(new Path(OUTPUT_PATH),true);
      }
      Job job = Job.getInstance(conf, "qq friend recommended");

      job.setJarByClass(FindFriends.class);

      FileInputFormat.addInputPath(job, new Path(INPUT_PATH));
      job.setMapperClass(MyMapper.class);
      job.setReducerClass(MyReudcer.class);

      job.setOutputKeyClass(Text.class);
      job.setOutputValueClass(Text.class);
      FileOutputFormat.setOutputPath(job, new Path(OUTPUT_PATH));

      //提交作业
      System.exit(job.waitForCompletion(true) ? 0 : 1);

}

public static class MyMapper extends Mapper<LongWritable,Text,Text,Text>{
      @Override
      protected void map(LongWritable k1, Text v1, Context context)
            throws IOException, InterruptedException {
         String line = v1.toString();
         String[] ss = line.split("\\s+");

         context.write(new Text(ss[0]), new Text(ss[1]));
         context.write(new Text(ss[1]), new Text(ss[0]));
      }
}

public static class MyReudcer extends Reducer<Text, Text, Text, Text>{
      @Override
      protected void reduce(Text k2, Iterable<Text> v2s,Context context)
            throws IOException, InterruptedException {
         Set<String> set = new HashSet<String>();
         for (Text v2:v2s) {
            set.add(v2.toString());
         }
         if (set.size()>1) {
            for (Iterator i = set.iterator();i.hasNext();) {

                  String qqName = (String)i.next();
                  for (Iterator j = set.iterator();j.hasNext();){
                     String otherqqName = (String)j.next();
                     if(!qqName.equals(otherqqName)){
                        context.write(new Text(qqName), new Text(otherqqName));
                     }
                  }
            }
         }
      }

}
}[/mw_shl_code]

输出结果：

翼宇轩