hadoop共同好友

以下是Hadoop求共同好友的步骤：

1. 将每个人的好友列表作为value，以人名为key，将数据存储在Hadoop的HDFS中。
2. 编写MapReduce程序，将每个人的好友列表作为value，以人名对为key，输出<key, value>对，其中key为两个人名的组合，value为这两个人的好友列表。
3. 编写第二个MapReduce程序，将第一步输出的<key, value>对作为输入，对于每个<key, value>对，将value中的好友列表取交集，输出<key, value>对，其中key为两个人名的组合，value为这两个人的共同好友列表。

下面是第一个MapReduce程序的代码：

public class FriendMapper extends Mapper<LongWritable, Text, Text, Text> {
    private Text outputKey = new Text();
    private Text outputValue = new Text();

    public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String[] line = value.toString().split(":");
        String person = line[0];
        String[] friends = line[1].split(",");
        for (String friend : friends) {
            String pair = getPair(person, friend);
            outputKey.set(pair);
            outputValue.set(line[1]);
            context.write(outputKey, outputValue);
        }
    }

    private String getPair(String person, String friend) {
        return person.compareTo(friend) < 0 ? person + "," + friend : friend + "," + person;
    }
}

下面是第二个MapReduce程序的代码：

public class FriendReducer extends Reducer<Text, Text, Text, Text> {
    private Text outputValue = new Text();

    public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        List<String> friendLists = new ArrayList<>();
        for (Text value : values) {
            friendLists.add(value.toString());
        }
        List<String> commonFriends = getCommonFriends(friendLists);
        outputValue.set(StringUtils.join(commonFriends, ","));
        context.write(key, outputValue);
    }

    private List<String> getCommonFriends(List<String> friendLists) {
        List<String> commonFriends = new ArrayList<>();
        Map<String, Integer> friendCount = new HashMap<>();
        for (String friendList : friendLists) {
            String[] friends = friendList.split(",");
            for (String friend : friends) {
                if (friendCount.containsKey(friend)) {
                    friendCount.put(friend, friendCount.get(friend) + 1);
                } else {
                    friendCount.put(friend, 1);
                }
            }
        }
        for (Map.Entry<String, Integer> entry : friendCount.entrySet()) {
            if (entry.getValue() == friendLists.size()) {
                commonFriends.add(entry.getKey());
            }
        }
        return commonFriends;
    }
}