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

Question

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

#!/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?