使用mrjob实现MapReduce时遇TypeError:无法解包非可迭代float对象

我正在通过简单示例学习MapReduce和mrjob，目标是计算所有数字的对数之和，再用数字总数除以该求和结果。

编写的代码如下（mrMedian.py）：

# mrMedian.py

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

import math

class MrMedian(MRJob):

    def __init__(self, *args, **kwargs):
        super(MrMedian, self).__init__(*args, **kwargs)
        self.inCount = 0
        self.inLogSum = 0.0

    #increment the count of elements and add the 
    # logarithm of the current number to the summation
    def map(self, key, val):
        inVal = float(val)
        self.inCount += 1
        self.inLogSum += math.log(inVal)

    # return the count and summation after all numbers are processed     
    def map_final(self):
        yield (1, [self.inCount, self.inLogSum])

    # aggregate the count and summation values and yield the result
    def reduce(self, key, packedValues):
        cumLogSum=1.0
        cumN=0

        for valArr in packedValues:
            nj = int(valArr[0])
            cumN += nj        
            cumLogSum += float(valArr[1])
        
        median = cumN/cumLogSum
        yield (median)

    # define mapper and reducer
    def steps(self):
        return ([
            MRStep(mapper=self.map, reducer=self.reduce, mapper_final=self.map_final)
        ])
    
# to run:
# python MrMedian.py < inputFile.txt     
if __name__ == '__main__':
    MrMedian.run()

运行时出现以下错误：

(venv) shahriar@Lenovo:/media/shahriar/01D779182B58B9D0$ python mrMedian.py < inputFile.txt > outFile.txt No configs found; falling back on auto-configuration No configs specified for inline runner Creating temp directory /tmp/mrMedian.shahriar.20221113.152412.029427 Running step 1 of 1... reading from STDIN

Error while reading from /tmp/mrMedian.shahriar.20221113.152412.029427/step/000/reducer/00000/input:

Traceback (most recent call last):   
    File "/media/shahriar/01D779182B58B9D0/assignment2/mrMedian.py", line 43, in <module>
    MrMedian.run()   
    File "/media/shahriar/01D779182B58B9D0/venv/lib/python3.10/site-packages/mrjob/job.py", line 616, in run
    cls().execute()   
    File "/media/shahriar/01D779182B58B9D0/venv/lib/python3.10/site-packages/mrjob/job.py", line 687, in execute
    self.run_job()   
    File "/media/shahriar/01D779182B58B9D0/venv/lib/python3.10/site-packages/mrjob/job.py", line 636, in run_job
    runner.run()   
    File "/media/shahriar/01D779182B58B9D0/venv/lib/python3.10/site-packages/mrjob/runner.py", line 503, in run
    self._run()   
    File "/media/shahriar/01D779182B58B9D0/venv/lib/python3.10/site-packages/mrjob/sim.py", line 161, in _run
    self._run_step(step, step_num)   
    File "/media/shahriar/01D779182B58B9D0/venv/lib/python3.10/site-packages/mrjob/sim.py", line 170, in _run_step
    self._run_streaming_step(step, step_num)   
    File "/media/shahriar/01D779182B58B9D0/venv/lib/python3.10/site-packages/mrjob/sim.py", line 187, in _run_streaming_step
    self._run_reducers(step_num, num_reducer_tasks)   
    File "/media/shahriar/01D779182B58B9D0/venv/lib/python3.10/site-packages/mrjob/sim.py", line 289, in _run_reducers
    self._run_multiple(   
    File "/media/shahriar/01D779182B58B9D0/venv/lib/python3.10/site-packages/mrjob/sim.py", line 130, in _run_multiple
    func()   
    File "/media/shahriar/01D779182B58B9D0/venv/lib/python3.10/site-packages/mrjob/sim.py", line 746, in _run_task
    invoke_task(   File "/media/shahriar/01D779182B58B9D0/venv/lib/python3.10/site-packages/mrjob/inline.py", line 133, in invoke_task
    task.execute()   
    File "/media/shahriar/01D779182B58B9D0/venv/lib/python3.10/site-packages/mrjob/job.py", line 681, in execute
    self.run_reducer(self.options.step_num)   
    File "/media/shahriar/01D779182B58B9D0/venv/lib/python3.10/site-packages/mrjob/job.py", line 795, in run_reducer
    for k, v in self.reduce_pairs(read_lines(), step_num=step_num):   
    File "/media/shahriar/01D779182B58B9D0/venv/lib/python3.10/site-packages/mrjob/job.py", line 866, in reduce_pairs
    for k, v in self._combine_or_reduce_pairs(pairs, 'reducer', step_num):   
    File "/media/shahriar/01D779182B58B9D0/venv/lib/python3.10/site-packages/mrjob/job.py", line 889, in _combine_or_reduce_pairs
    for k, v in task(key, values) or (): 
TypeError: cannot unpack non-iterable float object

经检查，map_final生成的reducer输入文件格式正常：

shahriar@Lenovo-:/tmp/mrMedian.shahriar.20221113.152412.029427/step/000/reducer/00000$ cat input 
1   [13, 78.5753201837955]
1   [13, 77.20894832945609]
1   [12, 75.70546637672973]
1   [12, 73.97942285230064]
1   [13, 78.7642193551817]
1   [13, 74.83203774429285]
1   [13, 72.28868623927899]
1   [11, 67.51370208632588]

即使注释reduce方法中的for循环，错误依然存在，寻求问题原因及解决办法。

核心错误原因 #

mrjob的Reducer要求yield的结果必须是键值对（二元组），代码中yield (median)只返回了单个float值，mrjob尝试将其解包成(k, v)格式时，就会抛出TypeError: cannot unpack non-iterable float object错误。

除此之外还有两个潜在问题：

1. cumLogSum初始值设为1.0会导致求和结果错误，应该初始化为0.0
2. mrjob会自动将mapper输出的列表转为字符串，Reducer接收到的packedValues是字符串格式，无法直接按列表索引取值

修正后的代码 #

# mrMedian.py

from mrjob.job import MRJob
from mrjob.step import MRStep
import math
import ast  # 新增：用于解析字符串化的列表

class MrMedian(MRJob):

    def __init__(self, *args, **kwargs):
        super(MrMedian, self).__init__(*args, **kwargs)
        self.inCount = 0
        self.inLogSum = 0.0

    def map(self, key, val):
        val = val.strip()
        if not val:  # 新增：跳过空行，避免无效输入
            return
        inVal = float(val)
        self.inCount += 1
        self.inLogSum += math.log(inVal)

    def map_final(self):
        yield (1, [self.inCount, self.inLogSum])

    def reduce(self, key, packedValues):
        cumLogSum = 0.0  # 修正：初始值从1.0改为0.0
        cumN = 0

        for valStr in packedValues:
            # 新增：将字符串解析为列表
            valArr = ast.literal_eval(valStr)
            nj = int(valArr[0])
            cumN += nj        
            cumLogSum += float(valArr[1])
        
        result = cumN / cumLogSum
        yield ("最终结果", result)  # 修正：返回键值对格式

    def steps(self):
        return ([
            MRStep(mapper=self.map, reducer=self.reduce, mapper_final=self.map_final)
        ])
    
if __name__ == '__main__':
    MrMedian.run()

关键改动说明 #

1. 修正Reducer输出格式：将yield (median)改为yield ("最终结果", result)，返回标准的键值对，符合mrjob的要求
2. 修复求和初始值：把cumLogSum的初始值从1.0改为0.0，避免初始值干扰对数总和的计算
3. 解析字符串化列表：使用ast.literal_eval()将Reducer接收到的字符串格式列表解析为Python列表，确保能正确获取count和logSum数值
4. 添加空行判断：在map方法中跳过空行，避免处理无效输入导致的错误

内容的提问来源于stack exchange，提问作者Shahriar.M