mrjob中add_file_arg()传递CSV报错的技术咨询

看起来你遇到的问题主要来自两个方面：命令行参数传递错误，以及对mrjob中选项访问和文件使用的小误解。让我们一步步解决它：

错误原因分析 #

你看到的 AttributeError: 'Values' object has no attribute 'database'，本质是因为你的命令行没有正确传递--database参数，导致self.options里根本没有database这个属性。你用了database=data/people_attributes.csv这种key=value的写法，但mrjob的命令行选项需要用--选项名 值的形式来传递。

修正步骤 #

1. 修正命令行执行命令 #

把原来的命令：

python3 calculate_people_scores.py --jobconf mapreduce.job.reduces=1 data/people_ids.csv database=data/people_attributes.csv

改成：

python3 calculate_people_scores.py --jobconf mapreduce.job.reduces=1 --database data/people_attributes.csv data/people_ids.csv

注意：--database data/people_attributes.csv要放在输入文件data/people_ids.csv前面，mrjob会把后面的路径当成任务的输入数据。

2. 优化代码：正确读取传入的CSV文件 #

仅仅打印路径还不够，我们需要在mapper中读取这个属性文件来匹配人员信息。推荐用mapper_init()方法来读取文件（每个mapper进程只读取一次，效率更高），而不是在每个mapper调用里重复读取：

import csv
from mrjob.job import MRJob

class MRPeopleScores(MRJob):
    def configure_args(self):
        super(MRPeopleScores, self).configure_args()
        # 给参数加上帮助信息，方便后续使用
        self.add_file_arg('--database', help='Path to the people attributes CSV file')

    def mapper_init(self):
        # 初始化时读取属性文件，存储到字典中方便快速查找
        self.person_attrs = {}
        with open(self.options.database, 'r') as f:
            reader = csv.DictReader(f)
            # 假设你的CSV有一个唯一标识列，比如'id'，作为字典的键
            for row in reader:
                person_id = row['id']
                self.person_attrs[person_id] = row

    def mapper(self, _, line):
        # 处理输入的people_ids.csv，每行是一个人员ID
        person_id = line.strip()
        # 查找对应的属性
        attrs = self.person_attrs.get(person_id)
        if attrs:
            # 输出你需要的键值对，比如人员ID和其属性
            yield person_id, attrs

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

为什么这样改？ #

• add_file_arg()的作用是告诉mrjob把指定的文件上传到每个任务节点的工作目录，这样mapper就能在本地访问它。
• mapper_init()是mrjob提供的钩子方法，会在mapper进程启动时执行一次，适合用来做初始化操作（比如读取配置文件、加载数据集），避免重复IO操作。
• 命令行参数的顺序很重要：mrjob会把所有不带--的路径当成任务的输入数据，所以要把--database这类选项放在输入文件前面。

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