Chaînage de plusieurs tâches MapReduce dans Hadoop

Dans de nombreuses situations réelles où vous appliquez MapReduce, les algorithmes finaux finissent par être plusieurs étapes MapReduce.

c'est-à-dire Map1, Reduce1, Map2, Reduce2, et ainsi de suite.

Vous avez donc la sortie de la dernière réduction qui est nécessaire comme entrée pour la carte suivante.

Les données intermédiaires sont quelque chose que vous ne souhaitez pas (en général) conserver une fois le pipeline terminé avec succès. De plus, parce que ces données intermédiaires sont en général une structure de données (comme une «carte» ou un «ensemble»), vous ne voulez pas mettre trop d'efforts dans l'écriture et la lecture de ces paires clé-valeur.

Quelle est la manière recommandée de le faire dans Hadoop?

Existe-t-il un exemple (simple) qui montre comment gérer correctement ces données intermédiaires, y compris le nettoyage par la suite?

Je pense que ce tutoriel sur le réseau de développeurs de Yahoo vous aidera avec ceci: chaîner des emplois

Vous utilisez le JobClient.runJob(). Le chemin de sortie des données du premier travail devient le chemin d'entrée vers votre deuxième travail. Celles-ci doivent être transmises en tant qu'arguments à vos travaux avec le code approprié pour les analyser et configurer les paramètres du travail.

Je pense que la méthode ci-dessus pourrait cependant être la manière dont l'API mapred maintenant plus ancienne l'a fait, mais elle devrait toujours fonctionner. Il y aura une méthode similaire dans la nouvelle API mapreduce mais je ne suis pas sûr de ce que c'est.

En ce qui concerne la suppression des données intermédiaires une fois le travail terminé, vous pouvez le faire dans votre code. La façon dont je l'ai fait avant utilise quelque chose comme:

FileSystem.delete(Path f, boolean recursive);

Où le chemin est l'emplacement sur HDFS des données. Vous devez vous assurer de ne supprimer ces données que lorsqu'aucune autre tâche ne les nécessite.

Il y a plusieurs façons de le faire.

(1) Travaux en cascade

Créez l'objet JobConf "job1" pour le premier travail et définissez tous les paramètres avec "input" comme répertoire d'entrée et "temp" comme répertoire de sortie. Exécutez ce travail:

JobClient.run(job1).

Juste en dessous, créez l'objet JobConf "job2" pour le deuxième travail et définissez tous les paramètres avec "temp" comme répertoire d'entrée et "sortie" comme répertoire de sortie. Exécutez ce travail:

JobClient.run(job2).

(2) Créez deux objets JobConf et définissez tous les paramètres qu'ils contiennent comme (1) sauf que vous n'utilisez pas JobClient.run.

Créez ensuite deux objets Job avec jobconfs comme paramètres:

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

À l'aide de l'objet jobControl, vous spécifiez les dépendances de travail, puis exécutez les travaux:

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

(3) Si vous avez besoin d'une structure un peu comme Map + | Réduire | Map *, vous pouvez utiliser les classes ChainMapper et ChainReducer fournies avec Hadoop version 0.19 et suivantes.

Il existe en fait plusieurs façons de procéder. Je vais me concentrer sur deux.

L'un est via Riffle ( http://github.com/cwensel/riffle ) une bibliothèque d'annotations pour identifier les choses dépendantes et les «exécuter» dans l'ordre de dépendance (topologique).

Ou vous pouvez utiliser une cascade (et MapReduceFlow) en cascade ( http://www.cascading.org/ ). Une future version prendra en charge les annotations Riffle, mais cela fonctionne très bien maintenant avec les tâches MR JobConf brutes.

Une variante à cela consiste à ne pas gérer du tout les travaux MR à la main, mais à développer votre application à l'aide de l'API en cascade. Ensuite, le JobConf et le chaînage des travaux sont gérés en interne via le planificateur en cascade et les classes Flow.

De cette façon, vous passez votre temps à vous concentrer sur votre problème, pas sur les mécanismes de gestion des tâches Hadoop, etc. Vous pouvez même superposer différents langages (comme clojure ou jruby) pour simplifier encore plus votre développement et vos applications. http://www.cascading.org/modules.html

J'ai fait le chaînage de tâches en utilisant les objets JobConf les uns après les autres. J'ai pris l'exemple de WordCount pour enchaîner les travaux. Un travail détermine combien de fois un mot est répété dans la sortie donnée. Le deuxième travail prend la première sortie de travail comme entrée et calcule le nombre total de mots dans l'entrée donnée. Vous trouverez ci-dessous le code qui doit être placé dans la classe 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);

La commande pour exécuter ces travaux est:

bin / pot hadoop TotalWords.

Nous devons donner le nom final des tâches à la commande. Dans le cas ci-dessus, il s'agit de TotalWords.

Vous pouvez exécuter la chaîne MR de la manière indiquée dans le code.

VEUILLEZ NOTER : Seul le code du pilote a été fourni

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 SÉQUENCE EST

( JOB1 ) MAP-> REDUCE-> ( JOB2 ) MAP
Cela a été fait pour que les clés soient triées, mais il existe d'autres moyens, comme l'utilisation d'un treemap.
Pourtant, je veux concentrer votre attention sur la façon dont les emplois ont été enchaînés! !
Je vous remercie

Vous pouvez utiliser oozie pour traiter vos tâches MapReduce. http://issues.apache.org/jira/browse/HADOOP-5303

Il existe des exemples dans le projet Apache Mahout qui enchaînent plusieurs tâches MapReduce. Un des exemples peut être trouvé à:

RecommenderJob.java

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

Nous pouvons utiliser la waitForCompletion(true)méthode du Job pour définir la dépendance entre le Job.

Dans mon scénario, j'avais 3 emplois qui dépendaient les uns des autres. Dans la classe de pilote, j'ai utilisé le code ci-dessous et cela fonctionne comme prévu.

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 nouvelle classe org.apache.hadoop.mapreduce.lib.chain.ChainMapper aide ce scénario

Bien qu'il existe des moteurs de flux de travail Hadoop basés sur des serveurs complexes, par exemple, oozie, j'ai une bibliothèque java simple qui permet l'exécution de plusieurs travaux Hadoop en tant que flux de travail. La configuration du travail et le flux de travail définissant la dépendance entre les travaux sont configurés dans un fichier JSON. Tout est configurable en externe et ne nécessite aucune modification de la carte existante, réduisez la mise en œuvre pour faire partie d'un flux de travail.

Les détails peuvent être trouvés ici. Le code source et le fichier jar sont disponibles dans github.

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

Pranab

Je pense que oozie aide les emplois conséquents à recevoir les intrants directement de l'emploi précédent. Cela évite les opérations d'E / S effectuées avec jobcontrol.

Si vous souhaitez enchaîner vos travaux par programmation, vous souhaiterez utiliser JobControl. L'utilisation est assez simple:

JobControl jobControl = new JobControl(name);

Après cela, vous ajoutez des instances ControlledJob. ControlledJob définit un travail avec ses dépendances, connectant ainsi automatiquement les entrées et les sorties pour s'adapter à une «chaîne» de travaux.

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

    jobControl.run();

démarre la chaîne. Vous voudrez mettre cela dans un fil speerate. Cela permet de vérifier l'état de votre chaîne pendant son exécution:

    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());
    }

Comme vous l'avez mentionné dans votre exigence selon laquelle vous voulez que l'o / p de MRJob1 soit l'i / p de MRJob2 et ainsi de suite, vous pouvez envisager d'utiliser le workflow oozie pour ce cas d'utilisation. Vous pouvez également envisager d'écrire vos données intermédiaires sur HDFS car elles seront utilisées par le prochain MRJob. Et une fois le travail terminé, vous pouvez nettoyer vos données intermédiaires.

<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"/>