python - ¿Cómo determinar específicamente la entrada para cada paso del mapa en MRJob?

hadoop mapreduce

Question

Estoy trabajando en un trabajo de reducción de mapas, que consta de varios pasos. Usando mrjob cada paso recibe salida de paso anterior. El problema es que no quiero que lo haga.

Lo que quiero es extraer algo de información y usarla en el segundo paso contra todas las entradas, y así sucesivamente. ¿Es posible hacer esto usando mrjob?

Nota : Como no quiero usar emr, esta pregunta no me sirve de mucho.

ACTUALIZACIÓN : Si no sería posible hacer esto en un solo trabajo, necesito hacerlo en dos trabajos separados. En este caso, ¿hay alguna manera de envolver estos dos trabajos y gestionar salidas intermedias, etc.?

Answer 1

Puedes usar Runners

Tendrá que definir los trabajos por separado y usar otro script de python para invocarlo.

from NumLines import NumLines
from WordsPerLine import WordsPerLine
import sys

intermediate = None

def firstJob(input_file):
    global intermediate
    mr_job = NumLines(args=[input_file])
    with mr_job.make_runner() as runner:
        runner.run()
        intermediate = runner.get_output_dir()

def secondJob(input_file):
    mr_job = WordsPerLine(args=[intermediate,input_file])
    with mr_job.make_runner() as runner:
        runner.run()

if __name__ == ''__main__'':
    firstJob(sys.argv[1]) 
    secondJob(sys.argv[1])

y puede ser invocado por:

python main_script.py input.txt