mrJob python mapReduce word_count.py

Question

Я только что начал использовать mrJob (mapReduce для python) и новичок в парадигме MapReduce, я хотел бы узнать следующее о учебнике word_count.py, который присутствует на сайте документации MRJob.

Документы говорят, что если мы создадим word_count.py и запустим его с помощью некоторого текстового файла, он будет вычислять и возвращать количество строк, символов и слов в текстовом файле. Вот код, который они используют для word_count.py:

from mrjob.job import MRJob 


class MRWordFrequencyCount(MRJob): 

    def mapper(self, _, line): 
     yield "chars", len(line) 
     yield "words", len(line.split()) 
     yield "lines", 1 

    def reducer(self, key, values): 
     yield key, sum(values) 


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

Здесь я понимаю, что мы расширим класс MRJob и переопределить картограф и редуктор методу. Но то, что я не получаю во время выполнения, мы выполняем, передавая весь текстовый файл следующим образом:

python word_count.py entire_text_file.txt 

так как делает картографа знать, как разобрать его на одну строку за один раз? В основном, мой вопрос заключается в том, каков будет вход в функцию mapper(), определенную выше? Будет ли это содержимое всего файла в целом или по одной строке за раз. И если это одна строка, какая часть кода MRJob заботится о поставке одной строки за раз в функцию mapper(). Надеюсь, что мой изначальный туманный вопрос был менее расплывчатым, но это меня полностью застопорило. Любая помощь будет оценена по достоинству.

Заранее благодарен!

Answer 1

хорошо, я думаю, лучший ответ RTFC: P

Если посмотреть в /usr/lib/python2.6/site-packages/mrjob/job.py (при установке mrjob с пип на python2 +0,6), вы увидите, как он точно читает строки из входного и выполнения картографа для каждой линии

def run_mapper(self, step_num=0): 
    ... 

    # pick input and output protocol 
    read_lines, write_line = self._wrap_protocols(step_num, 'mapper') 

    if mapper_init: 
     for out_key, out_value in mapper_init() or(): 
      write_line(out_key, out_value) 

    # run the mapper on each line 
    for key, value in read_lines(): 
     for out_key, out_value in mapper(key, value) or(): 
      write_line(out_key, out_value) 

    if mapper_final: 
     for out_key, out_value in mapper_final() or(): 
      write_line(out_key, out_value) 

Здесь определение read_lines

def _wrap_protocols(self, step_num, step_type): 
    """Pick the protocol classes to use for reading and writing 
    for the given step, and wrap them so that bad input and output 
    trigger a counter rather than an exception unless --strict-protocols 
    is set. 

    Returns a tuple of ``(read_lines, write_line)`` 

    ``read_lines()`` is a function that reads lines from input, decodes 
     them, and yields key, value pairs. 
    ``write_line()`` is a function that takes key and value as args, 
     encodes them, and writes a line to output. 

    :param step_num: which step to run (e.g. 0) 
    :param step_type: ``'mapper'``, ``'reducer'``, or ``'combiner'`` from 
         :py:mod:`mrjob.step` 
    """ 
    read, write = self.pick_protocols(step_num, step_type) 

    def read_lines(): 
     for line in self._read_input(): 
      try: 
       key, value = read(line.rstrip('\r\n')) 
       yield key, value 
      except Exception, e: 
       if self.options.strict_protocols: 
        raise 
       else: 
        self.increment_counter(
         'Undecodable input', e.__class__.__name__) 

    def write_line(key, value): 
     try: 
      print >> self.stdout, write(key, value) 
     except Exception, e: 
      if self.options.strict_protocols: 
       raise 
      else: 
       self.increment_counter(
        'Unencodable output', e.__class__.__name__) 

    return read_lines, write_line 

в конечном счете, вы можете прочитать read_input и read_file м это в /usr/lib/python2.6/site-packages/mrjob/util.py. надеюсь, это помогает