java - Hadoop M/R实现 “People You Might Know”友谊推荐

Question

如何通过查看两个人有多少个共同的 friend 来建立一个友谊推荐系统，并使用mapreduce工作将他们推荐为 friend ？有点像facebook或linkedin所做的事情，显示推荐人员列表，并按共同 friend 的数量对其进行排名。

Answer 1

该解决方案来自我的博客，我在项目中使用了此代码。

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

由于我不确定这是否是最佳解决方案，并且我也希望在stackoverflow中有一个文档，因此我在这里提出并回答了自己的问题。我希望获得社区的反馈。

最好的友谊推荐通常来自 friend 。关键思想是，如果两个人有很多共同的 friend ，但他们不是 friend ，则系统应建议他们彼此连接。
让我们假设友谊是无向的:如果A是B的 friend ，那么B也是A的 friend 。这是Facebook，Google +，Linkedin和几个社交网络中最常用的友谊系统。将其扩展到Twitter中使用的定向友谊系统并不困难；但是，在本文中，我们将重点关注无方向的案例。

输入数据将包含邻接列表，并以 的格式显示多行，其中 是唯一用户的唯一ID， 是用户列表，中间用 的 friend 的逗号。以下是输入示例。用户和用户之间的关系可以在图中更容易理解。

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的 friend ，但是用户0和用户4具有共同的 friend 1、2和3；用户0和用户5拥有共同的 friend 1。因此，我们建议将用户4和5推荐为用户0的 friend 。

推荐 friend 的输出将以以下格式给出。 <推荐给USER的 friend (共同 friend 的数量:[共同 friend 的ID，...])，...>。根据共同 friend 的数量对输出结果进行排序，并可以从图中进行验证。

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有 friend 1、2和3；结果，这对<1、2>，<2、1>，<2、3>，<3、2>，<1、3>和<3、1>具有用户0的共同 friend 。结果，我们可以发出 = <1，r = 2; m = 0>，<2，r = 1; m = 0>，<2，r = 3; m = 0> ...，其中r表示推荐的 friend ，m表示共同的 friend 。我们可以在缩减阶段汇总结果，并计算出用户和推荐用户之间有多少个共同的 friend 。但是，这种方法会引起问题。如果用户A和推荐用户B已经是 friend 怎么办？为了克服这个问题，我们可以在发射的值中添加另一个属性isFriend，如果我们知道他们已经是reduce阶段的 friend ，我们就不推荐该 friend 。在以下实现中，当他们已经成为 friend 时，将使用m = -1而不是使用额外的字段。

在输入数据中定义fromUser为 ，toUser为 之一，然后，算法可以由

map 阶段

Emit 。假设有n个toUser；那么我们将发出n条记录来描述fromUser和toUser已经是 friend 。请注意，它们已经是发出的键和r之间的 friend ，因此我们将m设置为-1。

Emit 用于toUser的toUser1和toUser2的所有可能组合，并且它们有共同的 friend fromUser。它将发出n(n-1)条记录。

在 map 阶段总共有n ^ 2个发射记录，其中n是 拥有的 friend 数量。

还原阶段

仅求和键和值r之间有多少个共同的 friend 。如果他们中有共同的 friend -1，由于他们已经是 friend ，因此我们不推荐您。

根据共同 friend 的数量对结果进行排序。

由于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));
                }
                }
            }
        }
    }

reducer 可以通过以下方式实现

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));
    }
}

其中，比较器在TreeMap中用于按共同好友数的降序对输出值进行排序。

欢迎任何评论和反馈。谢谢。