Python MapReduce两表关联问题：Reducer输出不符合预期

核心问题分析 #

你的Reducer代码存在三个关键逻辑错误，导致输出不符合预期：

1. 未处理记录顺序问题：MapReduce会将同一user_id的记录连续传入，但无法保证用户记录先于交易记录出现。原代码如果先收到交易记录，会把cur_location初始化为"-"，后续用户记录到来时错误地输出无效行，且无法更新交易记录的位置。
2. 输出逻辑错误：原代码在elif分支中只要是同一user_id就输出，但未区分当前行是用户记录还是交易记录，导致输出了大量product_id="-"的无效行。
3. 未缓存用户组记录：没有收集同一user_id的所有记录，无法确保先获取到用户位置再处理交易记录。

修正后的Reducer代码 #

#!/usr/bin/env python
import sys

def process_user_records(records):
    """处理单个用户的所有记录，提取位置并输出关联后的交易记录"""
    user_location = "-"
    transactions = []
    
    # 先遍历所有记录，收集用户位置和交易记录
    for rec in records:
        user_id, product_id, location = rec.split("\t")
        # 更新用户位置（优先取非"-"的有效值）
        if location != "-":
            user_location = location
        # 收集交易记录（product_id非"-"的行）
        if product_id != "-":
            transactions.append(product_id)
    
    # 输出所有关联后的交易记录
    for pid in transactions:
        print(f"{pid}\t{user_location}")

def main():
    last_user_id = None
    current_user_records = []
    
    for line in sys.stdin:
        line = line.strip()
        if not line:
            continue
        
        user_id, _, _ = line.split("\t")
        # 切换到新用户时，处理上一个用户的记录
        if user_id != last_user_id:
            if last_user_id is not None:
                process_user_records(current_user_records)
            # 重置缓存，开始收集新用户的记录
            last_user_id = user_id
            current_user_records = [line]
        else:
            # 同一用户，继续缓存记录
            current_user_records.append(line)
    
    # 处理最后一个用户的记录
    if current_user_records:
        process_user_records(current_user_records)

if __name__ == "__main__":
    main()

修正逻辑说明 #

1. 分组缓存记录：通过遍历输入流，将同一user_id的所有记录缓存到列表中，确保能完整获取该用户的所有数据。
2. 分离位置与交易记录：对每个用户的记录组，先提取有效的用户位置，再收集所有交易记录。
3. 关联输出：将每个交易记录与用户位置匹配后输出，确保每笔交易都能正确关联到对应的用户位置。

测试验证 #

使用你提供的Mapper输出作为输入，修正后的Reducer会输出：

1       US
1       GB
1       GB
2       FR
1       FR

输出内容完全符合预期（顺序因MapReduce排序规则可能略有不同，但关联关系正确）。

内容的提问来源于stack exchange，提问作者nkah