Hadoop M/R实施"People You Might Know"友情推荐

Question

如何建立一个友情推荐系统,通过查看两个共有多少朋友,并使用mapreduce工作推荐他们作为朋友？有点像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和几个社交网络中使用的最常见的友谊系统.将它扩展到推特中使用的定向友谊系统并不困难; 但是,我们将在整篇文章中关注无向的情况.

输入数据将包含邻接列表,并且具有<USER> <TAB> <FRIENDS>格式的多行,其中<USER>是唯一用户的唯一ID,<FRIENDS>是由以下分隔的用户列表:逗号谁是<USER>的朋友.以下是输入示例.在图中可以更容易地理解用户和用户之间的关系.

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的朋友.

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

定义fromUser是<USER>,并且toUser是输入数据中的<FRIENDS>之一,然后,算法可以由

地图阶段

1. Emit <fromUser,r = toUser; 所有toUser的m = -1>.假设有n个用户; 那么我们将发出n条记录来描述fromUser和toUser已经是朋友.请注意,它们已经是发出的键和r之间的朋友,因此我们将m设置为-1.
2. Emit <toUser1,r = toUser2; m = fromUser>对于toUser1的所有可能组合,以及toUser的toUser2,他们有共同的朋友,来自用户.它将发出n(n - 1)条记录.
3. 总的来说, 在地图阶段有n ^ 2个发射记录,其中n是<USER>拥有的朋友数.

减少阶段,

1. 只是总结他们在密钥和价值之间有多少共同的朋友.如果他们中的任何一方有共同朋友-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中用于按共享朋友数量的降序对输出值进行排序.

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