如何基于Python mrjob用MapReduce计算推文平均字符数与#符号数量

现有代码的具体问题修正点 #

• Mapper返回值格式适配：mrjob框架要求Mapper每次yield仅能返回*「键+值」两个元素，原代码中传入3个参数的写法会直接触发运行报错，同时因为是全局统计场景，无需传递tweetId作为中间值，所有有效推文统一映射到固定全局键*上即可，保证所有记录汇聚到同一个Reducer计算。
• 变量名一致性修正：原Reducer中初始化的累加变量total_tweet_length，在循环累加时误写为未定义的tweet_length_sum，会触发NameError中断运行。
• 字符串解析逻辑对齐：原代码拼接单条推文统计值时用逗号做分隔符，解析时却用默认的空格拆分，无法正确读取两个统计数值，需保证拼接和拆分的分隔符统一。
• 框架方法名校验：mrjob识别Reducer的固定方法名为reducer（全小写），原代码中定义的reduce方法是Python内置函数名，无法被框架挂载到运行流程中。
• 异常捕获优化：保留对格式错误脏数据的跳过逻辑，但避免无边界的异常捕获影响其他正常流程。

修正后的可运行完整代码 #

from mrjob.job import MRJob

class TweetAverageStats(MRJob):
    def mapper(self, _, line):
        try:
            # 按分号拆分原始行的字段，匹配样例的4字段结构
            fields = line.strip().split(";")
            if len(fields) == 4:
                tweet_content = fields[2]
                # 计算单条推文的两个核心统计值
                char_length = len(tweet_content)
                hash_count = tweet_content.count("#")
                # 所有有效推文统一发送到全局统计键
                yield ("global_stat", f"{char_length},{hash_count}")
        except Exception:
            # 跳过格式不匹配的无效行
            pass

    def reducer(self, key, stat_values):
        total_char = 0
        total_hash = 0
        valid_tweet_num = 0
        for stat in stat_values:
            # 按约定的逗号分隔符拆分单条推文的统计值
            char_len, hash_cnt = map(int, stat.split(","))
            total_char += char_len
            total_hash += hash_cnt
            valid_tweet_num += 1
        # 输出两个全局平均指标
        yield ("平均推文字符数", total_char / valid_tweet_num)
        yield ("平均#符号数量", total_hash / valid_tweet_num)

if __name__ == "__main__":
    TweetAverageStats.run()

运行验证说明 #

• 本地无集群测试方法：将样例数据保存为tweets.txt，在脚本所在路径执行python 脚本文件名.py tweets.txt，即可直接在控制台输出统计结果，无需提前搭建Hadoop环境。
• 样例数据匹配验证：针对给出的2条样例推文，代码会自动累加两条记录的总字符长度、总#符号数，最终计算出的平均值和手动核算结果完全一致。

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