Hadoop MapReduce添加Reducer后未获取期望聚合结果的问题咨询

已完成clients.csv与countries.csv的关联MapReduce任务，统计出各国家的“优质（bueno）”客户数量，现有步骤执行后得到结果：

["South Georgia and the South Sandwich Islands"]    1
["South Sudan"] 1
["Spain"]   3

尝试添加第三个Reducer实现两个需求：

1. 仅输出客户数量最多的单个结果（如3 Spain）
2. 输出所有客户数量最多的结果（如3 Spain和3 Guan）
   但添加的Reducer未实现聚合效果，仅数值运算生效，结果无变化。

问题原因 #

当前第二个Reducer的输出格式为(国家名称, 统计数量)，第三个Reducer直接接收该输出时，每个国家作为独立的键被分配到不同的Reducer实例中，每个实例仅处理单个国家的数值，调用max(values)只会返回该国家自身的统计数，无法实现全局聚合取最大值的效果。

解决方案 #

需要新增一个Mapper将第二个Reducer的输出进行格式转换，把所有数据归组到同一个键下，让第三个Reducer能获取全局的统计数据，再根据需求处理最大值。

步骤代码修改 #

首先更新steps方法，新增一个Mapper步骤：

def steps(self): 
    # ordenamos las operaciones para su ejecución.
    return [ 
        MRStep(mapper=self.mapper, reducer=self.reducer),  
        MRStep(mapper=self.mapper1, combiner=self.combiner_cuenta_palabras, reducer=self.reducer2),
        MRStep(mapper=self.mapper2, reducer=self.reducer3),            
    ]   

新增Mapper2转换数据格式 #

这个Mapper的作用是将所有统计结果归到同一个键下，确保所有数据进入同一个Reducer3实例：

def mapper2(self, country, count):
    # 将国家和数量封装为元组，统一以None作为键发送，同时清理国家名称的引号和括号
    yield None, (int(count), country.strip('[]"'))

需求1：仅输出单个数量最多的结果 #

实现Reducer3找出全局最大的统计项并输出：

def reducer3(self, _, values):
    # 元组比较会优先对比第一个元素（数量），直接取最大值即可
    max_count, max_country = max(values)
    yield max_count, max_country

需求2：输出所有数量最多的结果 #

先找出全局最大数量，再遍历所有项输出符合条件的结果：

def reducer3(self, _, values):
    value_list = list(values)
    # 提取所有数量，找出最大值
    max_count = max(item[0] for item in value_list)
    # 遍历输出所有数量等于最大值的项
    for count, country in value_list:
        if count == max_count:
            yield count, country

关键说明 #

• 必须通过Mapper2将数据归组到同一个键下，否则Reducer3无法获取全局数据进行聚合。
• 将count转换为int类型，避免字符串比较导致的错误（比如"10"会被字符串逻辑判定为小于"3"）。
• 清理国家名称的引号和括号是可选操作，目的是让输出格式更简洁。

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