python – 如何使用s3对象名作为MRJob映射器的输入,而不是s3对象本身？

我遗漏了一些关于Yelp的mrjob job库的明显信息.设置MRJob类几乎非常简单.在file或stdin上运行它也是如此.但是,如何在本地或在s3中将文件输入更改为作业,比如在s3存储桶中更改？

像这样的东西.假设我想计算我的S3存储桶中以字符串’foo’开头的所有对象：

import re

class MRCountS3Objects(MRJob):

    define mapper(self, _, botoS3Key):
        if re.match('^foo', botoS3Key.name):
            yield 'foo', 1

    define reduce(self, name, occurrences):
        yield name, sum(occurrences)

这是一个非常人为的例子,但你可能会得到我的漂移.如何告诉MRJob操作s3对象流,忽略对象的内容？我看到了S3Filesystem.get_s3_keys()method,它让我得到了我需要的流,但我不确定从那里开始.

解决方法:

想到至少一种方法来实现这一目标.您的MRJob具有可以分配给任何迭代器的stdin属性,然后您可以以编程方式运行该作业.例如,此代码应该适用于我的桶的密钥名称：

from mrjob.job import MRJob
from mrjob.emr import EMRJobRunner

class MRS3KeyProcessor(MRJob):
    # Do some MRJob stuff.
    ...

def s3_name_generator(bucket):
    """Generator that returns boto.s3.Key names.
    """
    # Could also use raw boto here.
    emr = EMRJobRunner()
    key_stream = emr.fs.get_s3_keys(bucket)
    for key in key_stream:
        yield key.name

def main():
    # The '-' argument signifies that we use stdin.
    mr_job = MRCountS3Objects(['--runner', 'inline', '-'])
    stdin = s3_name_generator('my-bucket')
    mr_job.stdin = stdin
    results = []
    with mr_job.make_runner() as runner:
        runner.run()
        for line in runner.stream_output():
            key, value = mr_job.parse_output_line(line)
            results.append((key, value))
    print(results)

if __name__ == '__main__':
    main()

标签：python,mapreduce,boto,elastic-map-reduce,mrjob
来源： https://codeday.me/bug/20190625/1284310.html