Encadenamiento de múltiples trabajos de MapReduce en Hadoop

En muchas situaciones de la vida real en las que aplica MapReduce, los algoritmos finales terminan siendo varios pasos de MapReduce.

es decir, Mapa1, Reducir1, Mapa2, Reducir2, etc.

Entonces tiene la salida de la última reducción que se necesita como entrada para el siguiente mapa.

Los datos intermedios son algo que (en general) no desea conservar una vez que la canalización se ha completado con éxito. Además, debido a que estos datos intermedios son, en general, una estructura de datos (como un 'mapa' o un 'conjunto'), no desea poner demasiado esfuerzo en escribir y leer estos pares clave-valor.

¿Cuál es la forma recomendada de hacerlo en Hadoop?

¿Hay un ejemplo (simple) que muestre cómo manejar estos datos intermedios de la manera correcta, incluida la limpieza posterior?

Creo que este tutorial en la red de desarrolladores de Yahoo te ayudará con esto: encadenando trabajos

Usas el JobClient.runJob(). La ruta de salida de los datos del primer trabajo se convierte en la ruta de entrada a su segundo trabajo. Estos deben pasarse como argumentos a sus trabajos con el código apropiado para analizarlos y configurar los parámetros para el trabajo.

Sin embargo, creo que el método anterior podría ser la forma en que lo hizo la API mapeada más antigua, pero aún así debería funcionar. Habrá un método similar en la nueva API mapreduce, pero no estoy seguro de qué es.

En cuanto a eliminar datos intermedios después de que un trabajo haya terminado, puede hacerlo en su código. La forma en que lo hice antes es usar algo como:

FileSystem.delete(Path f, boolean recursive);

Donde la ruta es la ubicación en HDFS de los datos. Debe asegurarse de eliminar solo estos datos una vez que ningún otro trabajo lo requiera.

Hay muchas formas de hacerlo.

(1) Trabajos en cascada

Cree el objeto JobConf "job1" para el primer trabajo y configure todos los parámetros con "input" como inputdirectory y "temp" como directorio de salida. Ejecute este trabajo:

JobClient.run(job1).

Inmediatamente debajo de él, cree el objeto JobConf "job2" para el segundo trabajo y configure todos los parámetros con "temp" como directorio de entrada y "output" como directorio de salida. Ejecute este trabajo:

JobClient.run(job2).

(2) Cree dos objetos JobConf y establezca todos los parámetros en ellos como (1), excepto que no use JobClient.run.

Luego cree dos objetos Job con jobconfs como parámetros:

Job job1=new Job(jobconf1); 
Job job2=new Job(jobconf2);

Usando el objeto jobControl, usted especifica las dependencias del trabajo y luego ejecuta los trabajos:

JobControl jbcntrl=new JobControl("jbcntrl");
jbcntrl.addJob(job1);
jbcntrl.addJob(job2);
job2.addDependingJob(job1);
jbcntrl.run();

(3) Si necesita una estructura similar a Map + | Reducir | Map *, puede usar las clases ChainMapper y ChainReducer que vienen con Hadoop versión 0.19 y posteriores.

En realidad, hay varias maneras de hacer esto. Me enfocaré en dos.

Una es a través de Riffle ( http://github.com/cwensel/riffle ) una biblioteca de anotaciones para identificar cosas dependientes y 'ejecutarlas' en orden de dependencia (topológico).

O puede usar una cascada (y MapReduceFlow) en cascada ( http://www.cascading.org/ ). Una versión futura admitirá anotaciones de Riffle, pero ahora funciona muy bien con trabajos de MR JobConf sin procesar.

Una variante de esto es no administrar los trabajos de MR a mano, sino desarrollar su aplicación utilizando la API en cascada. Luego, JobConf y el encadenamiento de trabajos se manejan internamente a través del planificador en cascada y las clases de flujo.

De esta manera, pasa su tiempo enfocándose en su problema, no en la mecánica de administrar los trabajos de Hadoop, etc. Incluso puede superponer diferentes idiomas (como clojure o jruby) para simplificar aún más su desarrollo y aplicaciones. http://www.cascading.org/modules.html

He realizado el encadenamiento de trabajos utilizando objetos JobConf uno tras otro. Tomé el ejemplo de WordCount para encadenar los trabajos. Un trabajo calcula cuántas veces se repite una palabra en la salida dada. El segundo trabajo toma la salida del primer trabajo como entrada y calcula el total de palabras en la entrada dada. A continuación se muestra el código que debe colocarse en la clase Driver.

    //First Job - Counts, how many times a word encountered in a given file 
    JobConf job1 = new JobConf(WordCount.class);
    job1.setJobName("WordCount");

    job1.setOutputKeyClass(Text.class);
    job1.setOutputValueClass(IntWritable.class);

    job1.setMapperClass(WordCountMapper.class);
    job1.setCombinerClass(WordCountReducer.class);
    job1.setReducerClass(WordCountReducer.class);

    job1.setInputFormat(TextInputFormat.class);
    job1.setOutputFormat(TextOutputFormat.class);

    //Ensure that a folder with the "input_data" exists on HDFS and contains the input files
    FileInputFormat.setInputPaths(job1, new Path("input_data"));

    //"first_job_output" contains data that how many times a word occurred in the given file
    //This will be the input to the second job. For second job, input data name should be
    //"first_job_output". 
    FileOutputFormat.setOutputPath(job1, new Path("first_job_output"));

    JobClient.runJob(job1);


    //Second Job - Counts total number of words in a given file

    JobConf job2 = new JobConf(TotalWords.class);
    job2.setJobName("TotalWords");

    job2.setOutputKeyClass(Text.class);
    job2.setOutputValueClass(IntWritable.class);

    job2.setMapperClass(TotalWordsMapper.class);
    job2.setCombinerClass(TotalWordsReducer.class);
    job2.setReducerClass(TotalWordsReducer.class);

    job2.setInputFormat(TextInputFormat.class);
    job2.setOutputFormat(TextOutputFormat.class);

    //Path name for this job should match first job's output path name
    FileInputFormat.setInputPaths(job2, new Path("first_job_output"));

    //This will contain the final output. If you want to send this jobs output
    //as input to third job, then third jobs input path name should be "second_job_output"
    //In this way, jobs can be chained, sending output one to other as input and get the
    //final output
    FileOutputFormat.setOutputPath(job2, new Path("second_job_output"));

    JobClient.runJob(job2);

El comando para ejecutar estos trabajos es:

bin / hadoop jar TotalWords.

Necesitamos dar el nombre de los trabajos finales para el comando. En el caso anterior, es TotalWords.

Puede ejecutar la cadena MR de la manera indicada en el código.

TENGA EN CUENTA : solo se ha proporcionado el código del controlador

public class WordCountSorting {
// here the word keys shall be sorted
      //let us write the wordcount logic first

      public static void main(String[] args)throws IOException,InterruptedException,ClassNotFoundException {
            //THE DRIVER CODE FOR MR CHAIN
            Configuration conf1=new Configuration();
            Job j1=Job.getInstance(conf1);
            j1.setJarByClass(WordCountSorting.class);
            j1.setMapperClass(MyMapper.class);
            j1.setReducerClass(MyReducer.class);

            j1.setMapOutputKeyClass(Text.class);
            j1.setMapOutputValueClass(IntWritable.class);
            j1.setOutputKeyClass(LongWritable.class);
            j1.setOutputValueClass(Text.class);
            Path outputPath=new Path("FirstMapper");
            FileInputFormat.addInputPath(j1,new Path(args[0]));
                  FileOutputFormat.setOutputPath(j1,outputPath);
                  outputPath.getFileSystem(conf1).delete(outputPath);
            j1.waitForCompletion(true);
                  Configuration conf2=new Configuration();
                  Job j2=Job.getInstance(conf2);
                  j2.setJarByClass(WordCountSorting.class);
                  j2.setMapperClass(MyMapper2.class);
                  j2.setNumReduceTasks(0);
                  j2.setOutputKeyClass(Text.class);
                  j2.setOutputValueClass(IntWritable.class);
                  Path outputPath1=new Path(args[1]);
                  FileInputFormat.addInputPath(j2, outputPath);
                  FileOutputFormat.setOutputPath(j2, outputPath1);
                  outputPath1.getFileSystem(conf2).delete(outputPath1, true);
                  System.exit(j2.waitForCompletion(true)?0:1);
      }

}

LA SECUENCIA ES

( TRABAJO1 ) MAPA- > REDUCIR-> ( TRABAJO2 ) MAPA
Esto se hizo para ordenar las claves, pero hay más formas de usar un mapa de árbol.
Sin embargo, ¡quiero centrar su atención en la forma en que se encadenaron los Trabajos! !
Gracias

Puede usar oozie para procesar en barch sus trabajos de MapReduce. http://issues.apache.org/jira/browse/HADOOP-5303

Hay ejemplos en el proyecto Apache Mahout que encadenan múltiples trabajos de MapReduce. Uno de los ejemplos se puede encontrar en:

RecomendadorJob.java

http://search-lucene.com/c/Mahout:/core/src/main/java/org/apache/mahout/cf/taste/hadoop/item/RecommenderJob.java%7C%7CRecommenderJob

Podemos hacer uso del waitForCompletion(true)método del Trabajo para definir la dependencia entre el trabajo.

En mi caso, tenía 3 trabajos que dependían unos de otros. En la clase de controlador utilicé el siguiente código y funciona como se esperaba.

public static void main(String[] args) throws Exception {
        // TODO Auto-generated method stub

        CCJobExecution ccJobExecution = new CCJobExecution();

        Job distanceTimeFraudJob = ccJobExecution.configureDistanceTimeFraud(new Configuration(),args[0], args[1]);
        Job spendingFraudJob = ccJobExecution.configureSpendingFraud(new Configuration(),args[0], args[1]);
        Job locationFraudJob = ccJobExecution.configureLocationFraud(new Configuration(),args[0], args[1]);

        System.out.println("****************Started Executing distanceTimeFraudJob ================");
        distanceTimeFraudJob.submit();
        if(distanceTimeFraudJob.waitForCompletion(true))
        {
            System.out.println("=================Completed DistanceTimeFraudJob================= ");
            System.out.println("=================Started Executing spendingFraudJob ================");
            spendingFraudJob.submit();
            if(spendingFraudJob.waitForCompletion(true))
            {
                System.out.println("=================Completed spendingFraudJob================= ");
                System.out.println("=================Started locationFraudJob================= ");
                locationFraudJob.submit();
                if(locationFraudJob.waitForCompletion(true))
                {
                    System.out.println("=================Completed locationFraudJob=================");
                }
            }
        }
    }

La nueva clase org.apache.hadoop.mapreduce.lib.chain.ChainMapper ayuda a este escenario

Aunque existen complejos motores de flujo de trabajo de Hadoop basados ​​en servidores, por ejemplo, oozie, tengo una biblioteca simple de Java que permite la ejecución de múltiples trabajos de Hadoop como flujo de trabajo. La configuración del trabajo y el flujo de trabajo que define la dependencia entre trabajos se configura en un archivo JSON. Todo es configurable externamente y no requiere ningún cambio en el mapa existente, lo que reduce la implementación para formar parte de un flujo de trabajo.

Detalles pueden ser encontrados aqui. El código fuente y el jar están disponibles en github.

http://pkghosh.wordpress.com/2011/05/22/hadoop-orchestration/

Pranab

Creo que oozie ayuda a los trabajos consiguientes a recibir las entradas directamente del trabajo anterior. Esto evita la operación de E / S realizada con control de trabajo.

Si desea encadenar sus trabajos mediante programación, deberá usar JobControl. El uso es bastante simple:

JobControl jobControl = new JobControl(name);

Después de eso, agrega instancias de ControlledJob. ControlledJob define un trabajo con sus dependencias, conectando automáticamente entradas y salidas para adaptarse a una "cadena" de trabajos.

    jobControl.add(new ControlledJob(job, Arrays.asList(controlledjob1, controlledjob2));

    jobControl.run();

comienza la cadena Querrás poner eso en un hilo de speerate. Esto permite verificar el estado de su cadena mientras se ejecuta:

    while (!jobControl.allFinished()) {
        System.out.println("Jobs in waiting state: " + jobControl.getWaitingJobList().size());
        System.out.println("Jobs in ready state: " + jobControl.getReadyJobsList().size());
        System.out.println("Jobs in running state: " + jobControl.getRunningJobList().size());
        List<ControlledJob> successfulJobList = jobControl.getSuccessfulJobList();
        System.out.println("Jobs in success state: " + successfulJobList.size());
        List<ControlledJob> failedJobList = jobControl.getFailedJobList();
        System.out.println("Jobs in failed state: " + failedJobList.size());
    }

Como ha mencionado en su requisito de que desea que o / p de MRJob1 sea el i / p de MRJob2, etc., puede considerar usar el flujo de trabajo de oozie para este caso de uso. También puede considerar escribir sus datos intermedios en HDFS, ya que serán utilizados por el próximo MRJob. Y una vez que se complete el trabajo, puede limpiar sus datos intermedios.

<start to="mr-action1"/>
<action name="mr-action1">
   <!-- action for MRJob1-->
   <!-- set output path = /tmp/intermediate/mr1-->
    <ok to="end"/>
    <error to="end"/>
</action>

<action name="mr-action2">
   <!-- action for MRJob2-->
   <!-- set input path = /tmp/intermediate/mr1-->
    <ok to="end"/>
    <error to="end"/>
</action>

<action name="success">
        <!-- action for success-->
    <ok to="end"/>
    <error to="end"/>
</action>

<action name="fail">
        <!-- action for fail-->
    <ok to="end"/>
    <error to="end"/>
</action>

<end name="end"/>