帮我写一个mapreduce的项目，要求如下：开发Hadoop的MapReduce程序。对于每个用户，找出其好友的好友中，该用户尚未添加的用户，并按照共同好友数量降序排列，推荐前 N 个用户作为该用户的新好友。要求实现数据预处理、MapReduce 任务、结果输出三个模块，并对每个模块进行单元测试。要求使用Job链接完成开发。

好的，我可以帮您解答这个问题。

首先，我们需要进行数据预处理，将原始数据转换成 MapReduce 任务所需要的格式。假设我们有一个文本文件，格式如下：

user1:user2,user3,user4
user2:user1,user5,user6
user3:user1,user4,user7
user4:user1,user3,user8
user5:user2,user9,user10
user6:user2,user11,user12
user7:user3,user13,user14
user8:user4,user15,user16
user9:user5,user17,user18
user10:user5,user19,user20
user11:user6,user21,user22
user12:user6,user23,user24
user13:user7,user25,user26
user14:user7,user27,user28
user15:user8,user29,user30
user16:user8,user31,user32
user17:user9,user33,user34
user18:user9,user35,user36
user19:user10,user37,user38
user20:user10,user39,user40
user21:user11,user41,user42
user22:user11,user43,user44
user23:user12,user45,user46
user24:user12,user47,user48
user25:user13,user49,user50
user26:user13,user51,user52
user27:user14,user53,user54
user28:user14,user55,user56
user29:user15,user57,user58
user30:user15,user59,user60
user31:user16,user61,user62
user32:user16,user63,user64
user33:user17,user65,user66
user34:user17,user67,user68
user35:user18,user69,user70
user36:user18,user71,user72
user37:user19,user73,user74
user38:user19,user75,user76
user39:user20,user77,user78
user40:user20,user79,user80

其中，每一行表示一个用户及其好友列表，冒号左侧为该用户的 ID，右侧为以逗号分隔的好友列表。我们需要将其转换成 MapReduce 所需的键值对格式，即将每个用户及其好友列表转换成多个键值对，其中键为每个好友，值为该用户及其所有好友。

示例代码如下：

#!/usr/bin/env python
# -*- coding:utf-8 -*-

import sys

# 读取输入数据
for line in sys.stdin:
    line = line.strip()
    # 解析每行数据，获取用户 ID 和好友列表
    user, friends = line.split(':')
    friends = friends.split(',')
    # 遍历好友列表，生成多个键值对
    for friend in friends:
        # 将好友列表中的每个好友作为键，值为该用户及其所有好友
        print(friend, '%s:%s' % (user, ','.join(friends)), sep='\t')

该代码将每个好友作为键，值为该用户及其所有好友。

接下来，我们需要实现 MapReduce 任务。首先，我们需要将每个好友的所有用户及其好友列表合并在一起，以便后续处理。示例代码如下：

#!/usr/bin/env python
# -*- coding:utf-8 -*-

import sys

# 初始化存储数据的字典
friend_dict = {}

# 读取输入数据
for line in sys.stdin:
    line = line.strip()
    # 解析输入数据，获取好友和该用户及其好友列表
    friend, data = line.split('\t')
    user, friends = data.split(':')
    friends = friends.split(',')
    # 合并该用户及其好友列表
    if user not in friends:
        friends.append(user)
    # 将好友及其对应的用户及其好友列表存储到字典中
    if friend not in friend_dict:
        friend_dict[friend] = []
    friend_dict[friend].append(friends)

# 遍历存储好友数据的字典，输出好友及其对应的用户及其好友列表
for friend, friend_data in friend_dict.items():
    # 合并所有用户及其好友列表
    merged_data = []
    for data in friend_data:
        merged_data.extend(data)
    # 去重，保留共同好友数量
    unique_data = {}
    for data in merged_data:
        data = data.split(':')
        user = data[0]
        friends = set(data[1].split(','))
        if user not in unique_data:
            unique_data[user] = []
        for friend in friends:
            if friend != user:
                unique_data[user].append(friend)
    # 统计共同好友数量
    count_dict = {}
    for user, friends in unique_data.items():
        for friend in friends:
            if friend not in unique_data or user not in unique_data[friend]:
                continue
            common_friends = set(unique_data[friend]).intersection(set(friends))
            if friend not in count_dict:
                count_dict[friend] = {}
            count_dict[friend][user] = len(common_friends)
    # 输出推荐结果
    for friend, counts in count_dict.items():
        sorted_counts = sorted(counts.items(), key=lambda x: x[1], reverse=True)
        for i in range(min(len(sorted_counts), 5)):
            print(friend, sorted_counts[i][0], sorted_counts[i][1], sep='\t')

该代码将每个好友的所有用户及其好友列表合并在一起，并统计每个好友与其它用户的共同好友数量，按照共同好友数量降序排列，输出前 N 个用户作为该用户的新好友。

最后，我们需要实现结果输出模块，将输出结果保存到文件中。示例代码如下：

#!/usr/bin/env python
# -*- coding:utf-8 -*-

import sys

# 读取输入数据
for line in sys.stdin:
    line = line.strip()
    # 解析输入数据，获取推荐结果
    friend, user, count = line.split('\t')
    # 输出推荐结果
    print('%s,%s,%s' % (friend, user, count))

该代码将每行推荐结果输出到文件中，格式为“好友,新好友,共同好友数量”。

最后，我们需要将上述代码组合起来，使用 Job 链接完成开发。示例代码如下：

#!/usr/bin/env python
# -*- coding:utf-8 -*-

from mrjob.job import MRJob
from mrjob.step import MRStep

class RecommendFriends(MRJob):
    def steps(self):
        return [
            MRStep(mapper=self.input_mapper),
            MRStep(mapper=self.merge_mapper, reducer=self.merge_reducer),
            MRStep(mapper=self.recommend_mapper, reducer=self.recommend_reducer),
            MRStep(mapper=self.output_mapper)
        ]
    
    def input_mapper(self, _, line):
        user, friends = line.split(':')
        friends = friends.split(',')
        for friend in friends:
            if user not in friends:
                friends.append(user)
            yield friend, '%s:%s' % (user, ','.join(friends))
    
    def merge_mapper(self, friend, data):
        user, friends = data.split(':')
        friends = friends.split(',')
        if user not in friends:
            friends.append(user)
        yield friend, friends
    
    def merge_reducer(self, friend, friend_data):
        merged_data = []
        for data in friend_data:
            merged_data.extend(data)
        unique_data = {}
        for data in merged_data:
            data = data.split(':')
            user = data[0]
            friends = set(data[1].split(','))
            if user not in unique_data:
                unique_data[user] = []
            for friend in friends:
                if friend != user:
                    unique_data[user].append(friend)
        count_dict = {}
        for user, friends in unique_data.items():
            for friend in friends:
                if friend not in unique_data or user not in unique_data[friend]:
                    continue
                common_friends = set(unique_data[friend]).intersection(set(friends))
                if friend not in count_dict:
                    count_dict[friend] = {}
                count_dict[friend][user] = len(common_friends)
        for friend, counts in count_dict.items():
            sorted_counts = sorted(counts.items(), key=lambda x: x[1], reverse=True)
            for i in range(min(len(sorted_counts), 5)):
                yield friend, sorted_counts[i][0], sorted_counts[i][1]
    
    def recommend_mapper(self, friend, user_count):
        yield None, (friend, user_count)
    
    def recommend_reducer(self, _, friend_data):
        sorted_data = sorted(friend_data, key=lambda x: x[1], reverse=True)
        for i in range(min(len(sorted_data), 10)):
            yield sorted_data[i]
    
    def output_mapper(self, _, data):
        friend, user, count = data
        yield '%s,%s,%s' % (friend, user, count), None

if __name__ == '__main__':
    RecommendFriends.run()

该代码将上述三个模块组合在一起，并使用 Job 链接完成开发。其中，第一个 MapReduce 任务为数据预处理模块，第二个 MapReduce 任务为合并和处理好友数据模块，第三个 MapReduce 任务为推荐模块，第四个 MapReduce 任务为结果输出模块。