Hadoop M/R实现“你可能认识的人”友谊推荐

Question

如何通过查看两人有多少共同的朋友来建立友谊推荐系统，并使用mapreduce job将他们推荐为朋友？有点像facebook或linkedin的做法，显示推荐人的列表，并根据共同好友的数量对他们进行排名。

Answer 1

这个解决方案取自我的博客，我在项目中使用了这段代码。

完整版本，请参阅https://www.dbtsai.com/blog/hadoop-mr-to-implement-people-you-might-know-friendship-recommendation/

由于我不确定这是否是一个最佳解决方案，而且我也想有一个stackoverflow文档，所以我在这里询问并回答了我自己的问题。我寻求社区的反馈。

最好的友情推荐往往来自朋友。关键的想法是，如果两个人有很多共同的朋友，但他们不是朋友，那么系统应该建议他们相互联系。让我们假设友谊是无方向的：如果A是B的朋友，那么B也是A的朋友。这是Facebook、Google+、Linkedin和几个社交网络中最常见的友谊系统。将其扩展到twitter中使用的定向友谊系统并不困难；然而，在这篇文章中，我们将重点讨论无向案例。

输入数据将包含邻接列表并且具有<用户>lt；TAB>lt；朋友>，其中<用户>是用于唯一用户的唯一ID，并且<朋友>是由逗号分隔的用户的列表，这些用户是<用户>。以下是一个输入示例。在图中可以更容易地理解用户和用户之间的关系。

1    0,2,3,4,5
2    0,1,4
3    0,1,4
4    1,2,3
5    1,6
6    5

在图中，您可以看到用户0不是用户4和5的朋友，但用户0和用户4有共同的朋友1、2和3；用户0和用户5具有共同的朋友1。因此，我们希望推荐用户4和5作为用户0的好友。

输出推荐的好友将以以下格式给出。<用户>lt；TAB>lt；向用户推荐好友（共同好友数：[共同好友id，…]），…>。输出结果根据共同好友的数量进行排序，可以从图中进行验证。

0    4 (3: [3, 1, 2]),5 (1: [1])
1    6 (1: [5])
2    3 (3: [1, 4, 0]),5 (1: [1])
3    2 (3: [4, 0, 1]),5 (1: [1])
4    0 (3: [2, 3, 1]),5 (1: [1])
5    0 (1: [1]),2 (1: [1]),3 (1: [1]),4 (1: [1])
6    1 (1: [5])

现在，让我们把这个问题放在单个M/R作业中。用户0有朋友1、2和3；结果；1、2>lt；2、1>lt；2、3>lt；3、2>lt；1、3>，并且<3、1>拥有用户0的共同好友。结果，我们可以发射<键，值>=<1，r＝2；m＝0>，<2，r＝1；m＝0>，<2，r=3；m＝0>。。。，其中r表示推荐的朋友，m表示共同的朋友。我们可以在减少阶段汇总结果，并计算用户和推荐用户之间有多少共同的朋友。然而，这种方法会造成问题。如果用户A和推荐的用户B已经是朋友了怎么办？为了克服这个问题，我们可以在发出的值中添加另一个属性isFriend，如果我们知道他们已经是reduce阶段的朋友，我们就不推荐这个朋友。在下面的实现中，当他们已经是朋友时，使用m=-1，而不是使用额外的字段。

定义fromUser是<USER>，toUser是<朋友>在输入数据中，然后，算法可以由给出

映射阶段

1. 发射<fromUser，r=toUser；m＝-1>所有toUser。假设有ntoUser；然后我们将发出n记录，用于描述fromUser和toUser已经是朋友。请注意，它们已经是已发出密钥和r之间的好友，因此我们将m设置为-1
2. 发射<toUser1，r=toUser2；m=fromUser>对于toUser1和toUser2从toUser的所有可能组合，他们有共同的朋友fromUser。它将发出n（n-1）个记录
3. 总共，在映射阶段有n^2个发射记录，其中n是好友数量<用户>有

减少相位，

1. 只是总结了他们之间有多少共同的朋友，在键和r之间的值。如果他们中的任何人有共同的朋友-1，我们不会推荐，因为他们已经是朋友了
2. 根据共同好友的数量对结果进行排序

因为发出的值不是hadoop中的基元数据类型，所以我们必须自定义一个新的可写类型作为以下代码。

static public class FriendCountWritable implements Writable {
    public Long user;
    public Long mutualFriend;
    public FriendCountWritable(Long user, Long mutualFriend) {
        this.user = user;
        this.mutualFriend = mutualFriend;
    }
    public FriendCountWritable() {
        this(-1L, -1L);
    }
    @Override
    public void write(DataOutput out) throws IOException {
        out.writeLong(user);
        out.writeLong(mutualFriend);
    }
    @Override
    public void readFields(DataInput in) throws IOException {
        user = in.readLong();
        mutualFriend = in.readLong();
    }
    @Override
    public String toString() {
        return " toUser: "
                + Long.toString(user) + " mutualFriend: "
                + Long.toString(mutualFriend);
    }
}

映射器可以通过实现

public static class Map extends Mapper<LongWritable, Text, LongWritable, FriendCountWritable> {
    private Text word = new Text();
    @Override
    public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String line[] = value.toString().split("t");
        Long fromUser = Long.parseLong(line[0]);
        List toUsers = new ArrayList();
        if (line.length == 2) {
            StringTokenizer tokenizer = new StringTokenizer(line[1], ",");
            while (tokenizer.hasMoreTokens()) {
                Long toUser = Long.parseLong(tokenizer.nextToken());
                toUsers.add(toUser);
                context.write(new LongWritable(fromUser),
                        new FriendCountWritable(toUser, -1L));
            }
            for (int i = 0; i < toUsers.size(); i++) {
                for (int j = i + 1; j < toUsers.size(); j++) {
                    context.write(new LongWritable(toUsers.get(i)),
                            new FriendCountWritable((toUsers.get(j)), fromUser));
                    context.write(new LongWritable(toUsers.get(j)),
                            new FriendCountWritable((toUsers.get(i)), fromUser));
                }
                }
            }
        }
    }

减速器可通过实现

public static class Reduce extends Reducer<LongWritable, FriendCountWritable, LongWritable, Text> {
    @Override
    public void reduce(LongWritable key, Iterable values, Context context)
            throws IOException, InterruptedException {
        // key is the recommended friend, and value is the list of mutual friends
        final java.util.Map<Long, List> mutualFriends = new HashMap<Long, List>();
        for (FriendCountWritable val : values) {
            final Boolean isAlreadyFriend = (val.mutualFriend == -1);
            final Long toUser = val.user;
            final Long mutualFriend = val.mutualFriend;
            if (mutualFriends.containsKey(toUser)) {
                if (isAlreadyFriend) {
                    mutualFriends.put(toUser, null);
                } else if (mutualFriends.get(toUser) != null) {
                    mutualFriends.get(toUser).add(mutualFriend);
                }
            } else {
                if (!isAlreadyFriend) {
                    mutualFriends.put(toUser, new ArrayList() {
                        {
                            add(mutualFriend);
                        }
                    });
                } else {
                    mutualFriends.put(toUser, null);
                }
            }
        }
        java.util.SortedMap<Long, List> sortedMutualFriends = new TreeMap<Long, List>(new Comparator() {
            @Override
            public int compare(Long key1, Long key2) {
                Integer v1 = mutualFriends.get(key1).size();
                Integer v2 = mutualFriends.get(key2).size();
                if (v1 > v2) {
                    return -1;
                } else if (v1.equals(v2) && key1 < key2) {
                    return -1;
                } else {
                    return 1;
                }
            }
        });
        for (java.util.Map.Entry<Long, List> entry : mutualFriends.entrySet()) {
            if (entry.getValue() != null) {
                sortedMutualFriends.put(entry.getKey(), entry.getValue());
            }
        }
        Integer i = 0;
        String output = "";
        for (java.util.Map.Entry<Long, List> entry : sortedMutualFriends.entrySet()) {
            if (i == 0) {
                output = entry.getKey().toString() + " (" + entry.getValue().size() + ": " + entry.getValue() + ")";
            } else {
                output += "," + entry.getKey().toString() + " (" + entry.getValue().size() + ": " + entry.getValue() + ")";
            }
            ++i;
        }
        context.write(key, new Text(output));
    }
}

其中Comparator在TreeMap中用于按照共同好友数量的递减顺序对输出值进行排序。

欢迎任何意见和反馈。谢谢