java - procesamiento - salida de un programa mapreduce como entrada a otro programa mapreduce

mapreduce google

Question

Estoy intentando un ejemplo simple, en el que el resultado de un trabajo de MapReduce debería ser la entrada de otro trabajo de MapReduce.

El flujo debería ser así: Mapper1 --> Reducer1 --> Mapper2 --> Reducer2 (La salida de Mapper1 debe ser la entrada de Reducer 1. La salida de Reducer1 debe ser la entrada de Mapper2. La salida de Mapper2 debe ser la entrada de Reducer2. La salida de Reducer2 debe almacenarse en el archivo de salida).

¿Cómo puedo agregar múltiples Mappers y Reductores a mi programa para que el flujo se mantenga como el anterior?

¿Necesito usar Chain Mappers o Chain Reducers? Si es así, ¿cómo puedo usarlos?

Answer 1

Necesita implementar dos trabajos MapReduce separados para eso. El resultado del primer trabajo debe escribirse en algún almacenamiento persistente (como HDFS) y el segundo trabajo lo leerá. El SequenceOutputFormat / InputFormat a menudo se usa para eso. Ambos trabajos de MapReduce se pueden ejecutar desde el mismo programa de controlador.

Answer 2

Supongo que lo que estás buscando es ControlledJob y JobControl. Se adapta muy bien a su propósito. En una sola clase de controlador puede construir múltiples trabajos que tienen dependencias entre sí. El siguiente código puede ayudarte a entender.

    Job jobOne = Job(jobOneConf, "Job-1");
    FileInputFormat.addInputPath(jobOne, jobOneInput);
    FileOutputFormat.setOutputPath(jobOne, jobOneOutput);
    ControlledJob jobOneControl = new ControlledJob(jobOneConf);
    jobOneControl.setJob(jobOne);

    Job jobTwo = Job(jobTwoConf, "Job-2");
    FileInputFormat.addInputPath(jobTwo, jobOneOutput); // here we set the job-1''s output as job-2''s input
    FileOutputFormat.setOutputPath(jobTwo, jobTwoOutput); // final output
    ControlledJob jobTwoControl = new ControlledJob(jobTwoConf);
    jobTwoControl.setJob(jobTwo);

    JobControl jobControl = new JobControl("job-control");
    jobControl.add(jobOneControl);
    jobControl.add(jobTwoControl);
    jobTwoControl.addDependingJob(jobOneControl); // this condition makes the job-2 wait until job-1 is done

    Thread jobControlThread = new Thread(jobControl);
    jobControlThread.start();
    jobControlThread.join(); 

    /* The jobControl.allFinished() can also be used to wait until all jobs are done */