基于MRJob编写MapReduce程序求CSV文件全局最大值

我来帮你搞定这个问题！要实现用MRJob找出CSV文件里所有数字的最大值，核心是在Map阶段正确解析每行里的所有数字（不管是逗号还是空格分隔），然后在Reduce阶段聚合找出最大值。下面是完整的解决方案：

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

class MRFindMaxNumber(MRJob):

    def steps(self):
        return [
            MRStep(mapper=self.mapper_extract_numbers,
                   reducer=self.reducer_find_max)
        ]

    def mapper_extract_numbers(self, _, line):
        # 把逗号替换成空格，统一所有分隔符
        cleaned_line = line.replace(',', ' ')
        # 按任意空白字符拆分（自动处理多个空格、制表符等情况）
        number_strings = cleaned_line.split()
        for num_str in number_strings:
            # 转换为整数（如果需要处理浮点数，把int改成float即可）
            try:
                number = int(num_str)
                # 用固定键'global_max'，确保所有数字都被送到同一个Reducer
                yield 'global_max', number
            except ValueError:
                # 跳过非数字的脏数据，提升程序健壮性
                continue

    def reducer_find_max(self, key, values):
        # 从所有传入的数值中提取最大值
        yield key, max(values)

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

• steps方法：定义MapReduce的执行流程，这里只需要一个Map步骤加一个Reduce步骤就能完成需求。
• mapper_extract_numbers：
  • 先把每行里的逗号替换成空格，让混合分隔的数字能统一按空白拆分。
  • 遍历拆分后的所有字符串，尝试转成数值类型，过滤掉非数字内容。
  • 每个有效数字都输出固定键的键值对，保证所有数字会被同一个Reducer处理。
• reducer_find_max：接收同一个键下的所有数值，直接用Python内置的max()函数找出最大值并输出。

假设你的输入文件numbers.csv内容如下：

1,1,1,1 2 3 4 5 6
7,8 9,10
15, 12 13

运行命令：

python your_script_name.py numbers.csv

最终输出结果会是：

"global_max"	15

原来的脚本只统计出现次数，说明它的Mapper逻辑可能只是按行或单一分隔符统计元素频次，没有把所有数字提取出来并传递给Reducer做最大值计算。我们的方案核心是将所有数字转换成可比较的数值类型，再通过Reducer聚合求最大值。

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