如何在MapReduce代码中使用emit_方法实现用户点击通话数据的输出示例？

假设我们有一个日志文件，其中包含用户点击通话的数据，每行格式如下：
user_id,action_time,phone_number
Map阶段
def map_user_clicks(line):
    # 解析日志行
    user_id, action_time, phone_number = line.split(',')
    
    # 输出键值对，其中键是用户ID，值是通话电话号码
    yield user_id, phone_number
Reduce阶段
def reduce_user_clicks(user_id, phone_numbers):
    # 将相同的用户ID的所有电话号码合并到一个列表中
    yield user_id, list(set(phone_numbers))
模拟MapReduce过程
def map_reduce_simulation(log_data):
    # 初始化Map结果
    map_results = []
    
    # 执行Map阶段
    for line in log_data:
        for user_id, phone_number in map_user_clicks(line):
            map_results.append((user_id, phone_number))
    
    # 初始化Reduce结果
    reduce_results = []
    
    # 执行Reduce阶段
    for user_id, phone_numbers in reduce_results:
        reduced_result = reduce_user_clicks(user_id, phone_numbers)
        for reduced_user_id, reduced_phone_numbers in reduced_result:
            reduce_results.append((reduced_user_id, reduced_phone_numbers))
    
    return reduce_results
示例日志数据
log_data = [
    "user1,20230401 12:00:00,1234567890",
    "user1,20230401 12:05:00,0987654321",
    "user2,20230401 12:10:00,1234567890",
    "user3,20230401 12:15:00,0987654321"
]
执行模拟
result = map_reduce_simulation(log_data)
打印结果
for user_id, phone_numbers in result:
    print(f"User ID: {user_id}, Phone Numbers: {phone_numbers}")

在这个示例中，我们首先定义了Map阶段的函数map_user_clicks，它接收一行日志数据，解析出用户ID和电话号码，然后生成键值对输出，我们定义了Reduce阶段的函数reduce_user_clicks，它接收用户ID和电话号码的列表，合并并去重电话号码，然后输出。