Как создать бегунка из гадюки?

Question

У меня есть следующие простые mrjob script, построчно считывающий большой файл, выполняет операцию над каждой строкой и печатает вывод:

#!/usr/bin/env python                                                                                                           

from mrjob.job import MRJob

class LineProcessor(MRJob):
    def mapper(self, _, line):
        yield (line.upper(), None) # toy example: mapper just uppercase the line

if __name__ == '__main__':
    # mr_job = LineProcessor(args=['-r', 'hadoop', '/path/to/input']) # error!
    mr_job = LineProcessor(args=['/path/to/input'])
    with mr_job.make_runner() as runner:
        runner.run() 
        for line in runner.stream_output():
            key, value = mr_job.parse_output_line(line)
            print key.encode('utf-8')  # don't care about value in my case

(Это просто игрушечный пример; обработка каждой строки в моем реальном случае обходится дорого, поэтому я хочу запускать распределенную.)

Он работает только как локальный процесс. Если я попытаюсь использовать '-r', 'hadoop' (см. комментарий выше) Я получаю следующую странную ошибку:

  File "mrjob/runner.py", line 727, in _get_steps
    'error getting step information: %s', stderr)
Exception: ('error getting step information: %s', 'Traceback (most recent call last):\n  File "script.py", line 11, in <module>\n    with mr_job.make_runner() as runner:\n  File "mrjob/job.py", line 515, in make_runner\n    " __main__, which doesn\'t work." % w)\nmrjob.job.UsageError: make_runner() was called with --steps. This probably means you tried to use it from __main__, which doesn\'t work.\n')

Как я могу на самом деле запустить его на Hadoop, то есть создать HadoopJobRunner?

Answer 1

Ты скучаешь

def steps(self):
        return [self.mr(
                          mapper_init = ...
                          mapper = self.mapper,
                          combiner = ...,
                          reducer = ...,
                  )]

в вашем LineProcessor?