寫在前面

首先我們來看看需求，以下是某部落格的好友列表資料，冒號前是一個使用者，冒號後是該使用者的所有好友（好友關係是單向的）：

A:B,C,D,F,E,O
B:A,C,E,K
C:F,A,D,I
D:A,E,F,L
E:B,C,D,M,L
F:A,B,C,D,E,O,M
G:A,C,D,E,F
H:A,C,D,E,O
I:A,O
J:B,O
K:A,C,D
L:D,E,F
M:E,F,G
O:A,H,I,J

程式碼

Mapper

public class FriendsMapper extends Mapper<LongWritable, Text, Text, Text> {
    Text k = new Text();
    Text v = new Text();

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        // A:B,C,D,F,E,O
        String line = value.toString();

        // 切割資料 變成: key:A value:B,C,D,F,E,O
        String[] fields = line.split(":");
        // 封裝k,v 寫出資料
        if (ArrayUtils.isNotEmpty(fields)) {
            k.set(fields[0]);
            v.set(fields[1]);
        }

        context.write(k, v);

    }
}
 

Reducer

public class FriendsReducer extends Reducer<Text, Text, Text, NullWritable> {
    private final Map<String, String> oldMap = new HashMap<>();
    private final Text k = new Text();

    @Override
    protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        // 將key與value封裝成map
        for (Text value : values) {
            oldMap.put(key.toString(), value.toString());
        }
    }

    /**
     * cleanup方法發生在所有reduce執行完之後，故這時候的oldMap已經填充了所有人的資料
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        // key:A value:B,C,D,F,E,O
        // 使用一個list儲存所有人的相同好友
        List<String> sameFriends = new ArrayList<>();
        // 獲取所有使用者集合的key
        Set<String> keySet = oldMap.keySet();

        // 開始迭代
        for (String keyOne:keySet) {

            for (String keyTwo:keySet) {
                // 獲取第一個人的好友
                List<String> firstCommonList = getCommonList(keyOne);

                // 獲取第二個人的好友
                List<String> secondCommonList = getCommonList(keyTwo);

                // 求二者的交集,儲存到firstCommonList
                firstCommonList.retainAll(secondCommonList);

                // 拼接要輸出的字串
                StringBuilder sb = new StringBuilder();
                for (String s : firstCommonList) {
                    sb.append(s).append(" ");
                }

                // 篩選掉沒有共同好友的人，即sb沒有拼接，為空串
                if (!StringUtils.equals("",sb.toString())) {
                    sameFriends.add(keyOne + "和" + keyTwo + "的共同好友為:" + sb.toString());
                }
            }
        }

        // 拼接完畢，輸出共同好友
        for (String sameFriend : sameFriends) {
            k.set(sameFriend);
            context.write(k, NullWritable.get());
        }
    }

    private List<String> getCommonList(String key) {
        String friend = oldMap.get(key);
        String[] common = friend.split(",");
        List<String> commonListTmp = Arrays.asList(common);
        return new ArrayList<>(commonListTmp);
    }
}
 

Driver

public class FriendDriver {

    public static void main(String[] args)  throws Exception{
        args = new String[]{"f:/hadoop/hadoopinput/friends.txt", "f:/hadoop/friendsouput"};

        Configuration conf = new Configuration();

        Job job = Job.getInstance(conf);
        job.setJarByClass(FriendDriver.class);

        job.setMapperClass(FriendsMapper.class);
        job.setReducerClass(FriendsReducer.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);

        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        boolean result = job.waitForCompletion(true);

        System.out.println(result ? 1 : 0);
    }
}
 

思路講解

首先我們考慮用hadoop中的MapReduce技術解決問題，MapReduce主要分為兩個階段：map階段和reduce階段。map階段負責聚集資料，reduce階段負責計算資料。

我們在map階段，先把資料按照冒號進行切割成如下格式：

key:A value:B,C,D,F,E,O

將其封裝成reduce階段的key和value，傳入到reduce階段。

在reduce階段，我們把所有的key和value儲存到一個Map鍵值對結構中，方便接下來的操作。

在Reduce的最後結束的cleanup階段，我們已經有了一個以使用者的ID(名字)為key，使用者擁有的好友為value的Map，我們對所有人的ID進行一個迴圈，依次比對每個ID擁有的好友，比如第一個人擁有的好友為集合A，第二個人擁有的好友為集合B，我們對集合A和B取交集，就能獲得共同好友了。之後我們對要輸出的結果進行一頓拼接，拼接出要輸出的結果。最後將其寫出即可。
最後輸出的結果為：

A和A的共同好友為:B C D F E O 
A和B的共同好友為:C E 
A和C的共同好友為:D F 
A和D的共同好友為:F E 
A和E的共同好友為:B C D 
A和F的共同好友為:B C D E O 
A和G的共同好友為:C D F E 
A和H的共同好友為:C D E O 
A和I的共同好友為:O 
A和J的共同好友為:B O 
A和K的共同好友為:C D 
A和L的共同好友為:D F E 
A和M的共同好友為:F E 
B和A的共同好友為:C E 
B和B的共同好友為:A C E K 
B和C的共同好友為:A 
B和D的共同好友為:A E 
B和E的共同好友為:C 
B和F的共同好友為:A C E 
B和G的共同好友為:A C E 
B和H的共同好友為:A C E 
B和I的共同好友為:A 
B和K的共同好友為:A C 
B和L的共同好友為:E 
B和M的共同好友為:E 
B和O的共同好友為:A 
C和A的共同好友為:F D 
C和B的共同好友為:A 
C和C的共同好友為:F A D I 
C和D的共同好友為:F A 
C和E的共同好友為:D 
C和F的共同好友為:A D 
C和G的共同好友為:F A D 
C和H的共同好友為:A D 
C和I的共同好友為:A 
C和K的共同好友為:A D 
C和L的共同好友為:F D 
C和M的共同好友為:F 
C和O的共同好友為:A I 
D和A的共同好友為:E F 
D和B的共同好友為:A E 
D和C的共同好友為:A F 
D和D的共同好友為:A E F L 
D和E的共同好友為:L 
D和F的共同好友為:A E 
D和G的共同好友為:A E F 
D和H的共同好友為:A E 
D和I的共同好友為:A 
D和K的共同好友為:A 
D和L的共同好友為:E F 
D和M的共同好友為:E F 
D和O的共同好友為:A 
E和A的共同好友為:B C D 
E和B的共同好友為:C 
E和C的共同好友為:D 
E和D的共同好友為:L 
E和E的共同好友為:B C D M L 
E和F的共同好友為:B C D M 
E和G的共同好友為:C D 
E和H的共同好友為:C D 
E和J的共同好友為:B 
E和K的共同好友為:C D 
E和L的共同好友為:D 
F和A的共同好友為:B C D E O 
F和B的共同好友為:A C E 
F和C的共同好友為:A D 
F和D的共同好友為:A E 
F和E的共同好友為:B C D M 
F和F的共同好友為:A B C D E O M 
F和G的共同好友為:A C D E 
F和H的共同好友為:A C D E O 
F和I的共同好友為:A O 
F和J的共同好友為:B O 
F和K的共同好友為:A C D 
F和L的共同好友為:D E 
F和M的共同好友為:E 
F和O的共同好友為:A 
G和A的共同好友為:C D E F 
G和B的共同好友為:A C E 
G和C的共同好友為:A D F 
G和D的共同好友為:A E F 
G和E的共同好友為:C D 
G和F的共同好友為:A C D E 
G和G的共同好友為:A C D E F 
G和H的共同好友為:A C D E 
G和I的共同好友為:A 
G和K的共同好友為:A C D 
G和L的共同好友為:D E F 
G和M的共同好友為:E F 
G和O的共同好友為:A 
H和A的共同好友為:C D E O 
H和B的共同好友為:A C E 
H和C的共同好友為:A D 
H和D的共同好友為:A E 
H和E的共同好友為:C D 
H和F的共同好友為:A C D E O 
H和G的共同好友為:A C D E 
H和H的共同好友為:A C D E O 
H和I的共同好友為:A O 
H和J的共同好友為:O 
H和K的共同好友為:A C D 
H和L的共同好友為:D E 
H和M的共同好友為:E 
H和O的共同好友為:A 
I和A的共同好友為:O 
I和B的共同好友為:A 
I和C的共同好友為:A 
I和D的共同好友為:A 
I和F的共同好友為:A O 
I和G的共同好友為:A 
I和H的共同好友為:A O 
I和I的共同好友為:A O 
I和J的共同好友為:O 
I和K的共同好友為:A 
I和O的共同好友為:A 
J和A的共同好友為:B O 
J和E的共同好友為:B 
J和F的共同好友為:B O 
J和H的共同好友為:O 
J和I的共同好友為:O 
J和J的共同好友為:B O 
K和A的共同好友為:C D 
K和B的共同好友為:A C 
K和C的共同好友為:A D 
K和D的共同好友為:A 
K和E的共同好友為:C D 
K和F的共同好友為:A C D 
K和G的共同好友為:A C D 
K和H的共同好友為:A C D 
K和I的共同好友為:A 
K和K的共同好友為:A C D 
K和L的共同好友為:D 
K和O的共同好友為:A 
L和A的共同好友為:D E F 
L和B的共同好友為:E 
L和C的共同好友為:D F 
L和D的共同好友為:E F 
L和E的共同好友為:D 
L和F的共同好友為:D E 
L和G的共同好友為:D E F 
L和H的共同好友為:D E 
L和K的共同好友為:D 
L和L的共同好友為:D E F 
L和M的共同好友為:E F 
M和A的共同好友為:E F 
M和B的共同好友為:E 
M和C的共同好友為:F 
M和D的共同好友為:E F 
M和F的共同好友為:E 
M和G的共同好友為:E F 
M和H的共同好友為:E 
M和L的共同好友為:E F 
M和M的共同好友為:E F G 
O和B的共同好友為:A 
O和C的共同好友為:A I 
O和D的共同好友為:A 
O和F的共同好友為:A 
O和G的共同好友為:A 
O和H的共同好友為:A 
O和I的共同好友為:A 
O和K的共同好友為:A 
O和O的共同好友為:A H I J 

總結

可能我解釋的不是很清楚，程式碼裡都有相應的註釋，這塊確實有點難理解，也不好解釋，如果需要的話就多看幾遍多理解吧。