Compiler et exécuter dynamiquement un travail Hadoop à partir d'un autre fichier Java

J'essaie d'écrire un fichier Java qui reçoit le code source d'un travail MapReduce, le compile dynamiquement et exécute le travail sur un cluster Hadoop. Pour y parvenir, j'ai écrit 3 méthodes appelées compile (), makeJAR () et run_Hadoop_Job (). Tout fonctionne bien avec la compilation et la création du fichier JAR. Cependant, lorsque le travail est soumis à Hadoop, dès le démarrage du travail, il rencontre un problème avec la recherche des classes de mappeur/réducteur requises et lève une ClassNotFoundException pour les deux le Mapper_Class et Reducer_Class *(java.lang.ClassNotFoundException: reza.rCloud.Mapper_Reducer_Classes$Mapper_Class.classe)* . Je sais qu'il devrait y avoir quelque chose de mal avec la façon dont j'ai référencé les classes de Mappeur/réducteur requises, mais je n'ai pas pu le comprendre après plusieurs. Toute aide / suggestion sur la façon de résoudre le problème est très appréciée.

En ce qui concerne les détails du projet: J'ai un fichier appelé "rCloud_test/src/reza/Mapper_Reducer_Classes.Java" qui contient le code source pour Mapper_Class et Reducer_Class. Ce fichier est finalement reçu pendant l'exécution mais pour l'instant j'y ai copié l'exemple Hadoop WordCount et je le stocke localement dans le même dossier que mon fichier de classe principal: rCloud_test/src/reza/Platform2.Java.

Ci-dessous vous pouvez voir la méthode main() de la Platform2.java, qui est ma classe principale pour ce projet:

    public static void main(String[] args){
    System.out.println("Code Execution Started");
    String className = "Mapper_Reducer_Classes";
    Platform2 myPlatform = new Platform2();

    //step 1: compile the received class file dynamically:
    boolean compResult = myPlatform.compile(className); 
    System.out.println(className + ".java compilation result: "+compResult);

    //step 2: make a JAR file out of the compiled file:
    if (compResult) {
        compResult  = myPlatform.makeJAR("jar_file", myPlatform.compilation_Output_Folder);
        System.out.println("JAR creation result: "+compResult);
    } 
    //step 3: Now let's run the Hadoop job:
    if (compResult) {       
        compResult = myPlatform.run_Hadoop_Job(className);
        System.out.println("Running on Hadoop result: "+compResult);
    }

La méthode qui me cause tous les problèmes est run_Hadoop_Job () qui est comme ci-dessous:

private boolean run_Hadoop_Job(String className){
try{
    System.out.println("*Starting to run the code on Hadoop...");
    String[] argsTemp = { "project_test/input", "project_test/output" };

    Configuration conf = new Configuration();
    conf.set("fs.default.name", "hdfs://localhost:54310");
    conf.set("mapred.job.tracker", "localhost:54311");

    conf.set("mapred.jar", jar_Output_Folder + "/jar_file"+".jar");

    conf.set("libjars", required_Execution_Classes);

    //THIS IS WHERE IT CAN'T FIND THE MENTIONED CLASSES, ALTHOUGH THEY EXIST BOTH ON DISK 
    // AND IN THE CREATED JAR FILE:??????
    System.out.println("Getting Mapper/Reducer package name: " + 
                        Mapper_Reducer_Classes.class.getName());
    conf.set("mapreduce.map.class", "reza.rCloud.Mapper_Reducer_Classes$Mapper_Class");
    conf.set("mapreduce.reduce.class", "reza.rCloud.Mapper_Reducer_Classes$Reducer_Class");

    Job job = new Job(conf, "Hadoop Example for dynamically and programmatically compiling-running a job");
    job.setJarByClass(Platform2.class);

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

    FileInputFormat.addInputPath(job, new Path(argsTemp[0]));
    FileSystem fs = FileSystem.get(conf);
    Path out = new Path(argsTemp[1]);
    fs.delete(out, true);
    FileOutputFormat.setOutputPath(job, new Path(argsTemp[1]));

    //job.submit();
    System.out.println("*and now submitting the job to Hadoop...");
    System.exit(job.waitForCompletion(true) ? 0 : 1); 
    System.out.println("Job Finished!");        
} catch (Exception e) {         
            System.out.println("****************Exception!" ); 
            e.printStackTrace();
            return false; 
    }
return true;
}

Si nécessaire, voici le code source de la méthode compile ():

private boolean compile(String className) {
    String fileToCompile =  JOB_FOLDER   + "/" +className+".java";
    JavaCompiler compiler = ToolProvider.getSystemJavaCompiler();       
    FileOutputStream errorStream = null;        
    try{
        errorStream = new FileOutputStream(JOB_FOLDER + "/logs/Errors.txt");
    } catch(FileNotFoundException e){ 
        //if problem creating the file, default wil be console 
    }   

    int compilationResult = 
            compiler.run(   null, null, errorStream, 
                            "-classpath", required_Compilation_Classes,
                            "-d", compilation_Output_Folder,
                            fileToCompile);
    if (compilationResult == 0) {
        //Compilation is successful:
        return true;
    } else {
        //Compilation Failed:
        return false;
    }   
}

Et le code source de la méthode makeJAR ():

private boolean makeJAR(String outputFileName, String inputDirectory) {
    Manifest manifest = new Manifest();
    manifest.getMainAttributes().put(Attributes.Name.MANIFEST_VERSION,
            "1.0");

    JarOutputStream target = null;                      
    try {       
        target = new JarOutputStream(new FileOutputStream(
                jar_Output_Folder+ "/"  
                + outputFileName+".jar"                      ), manifest);
        add(new File(inputDirectory), target);
    } catch (Exception e) { return false; }
    finally {
        if (target != null)
            try{
                target.close();
            } catch (Exception e) { return false; }
    }
    return true;
}

    private void add(File source, JarOutputStream target) throws IOException
{
  BufferedInputStream in = null;
  try
  {
    if (source.isDirectory())
    {
      String name = source.getPath().replace("\\", "/");
      if (!name.isEmpty())
      {
        if (!name.endsWith("/"))
          name += "/";
        JarEntry entry = new JarEntry(name);
        entry.setTime(source.lastModified());
        target.putNextEntry(entry);
        target.closeEntry();
      }
      for (File nestedFile: source.listFiles())
        add(nestedFile, target);
      return;
    }

    JarEntry entry = new JarEntry(source.getPath().replace("\\", "/"));
    entry.setTime(source.lastModified());
    target.putNextEntry(entry);
    in = new BufferedInputStream(new FileInputStream(source));

    byte[] buffer = new byte[1024];
    while (true)
    {
      int count = in.read(buffer);
      if (count == -1)
        break;
      target.write(buffer, 0, count);
    }
    target.closeEntry();
  }
  finally
  {
    if (in != null)
      in.close();
  }
}

Et enfin les paramètres fixes utilisés pour accéder aux fichiers:

private String JOB_FOLDER = "/Users/reza/My_Software/rCloud_test/src/reza/rCloud";
private String HADOOP_SOURCE_FOLDER = "/Users/reza/My_Software/hadoop-0.20.2";
private String required_Compilation_Classes = HADOOP_SOURCE_FOLDER + "/hadoop-0.20.2-core.jar";
private String required_Execution_Classes = required_Compilation_Classes + "," +
     "/Users/reza/My_Software/ActorFoundry_dist_ver/lib/commons-cli-1.1.jar," +
     "/Users/reza/My_Software/ActorFoundry_dist_ver/lib/commons-logging-1.1.1.jar";
public String compilation_Output_Folder = "/Users/reza/My_Software/rCloud_test/dyn_classes";
private String jar_Output_Folder = "/Users/reza/My_Software/rCloud_test/dyn_jar";

À la suite de l'exécution de la Platform2, la structure du projet sur le disque est la suivante:

RCloud_test/classes/reza/rCloud/Platform2.classe: contient la classe Platform2 rCloud_test/dyn_classes/reza/ rCloud / contient les classes pour Mapper_Reducer_Classes.classe, Mapper_Reducer_Classes$Mapper_Class.classe, et Mapper_Reducer_Classes$Reducer_Class.classe rCloud_test / dyn_jar / jar_file.jar contient le fichier jar créé

REVSED: voici le code source pour le rCloud_test/src/reza/rCloud/Mapper_Reducer_Classes.java:

package reza.rCloud;

import java.io.IOException;
import java.lang.InterruptedException;
import java.util.StringTokenizer;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

public class Mapper_Reducer_Classes {
/**
 * The map class of WordCount.
 */
public static class Mapper_Class
    extends Mapper<Object, Text, Text, IntWritable> {

    private final static IntWritable one = new IntWritable(1);
    private Text word = new Text();

    public void map(Object key, Text value, Context context)
        throws IOException, InterruptedException {
        StringTokenizer itr = new StringTokenizer(value.toString());
        while (itr.hasMoreTokens()) {
            word.set(itr.nextToken());
            context.write(word, one);
        }
    }
}
/**
 * The reducer class of WordCount
 */
public static class Reducer_Class
    extends Reducer<Text, IntWritable, Text, IntWritable> {
    public void reduce(Text key, Iterable<IntWritable> values, Context context)
        throws IOException, InterruptedException {
        int sum = 0;
        for (IntWritable value : values) {
            sum += value.get();
        }
        context.write(key, new IntWritable(sum));
    }
}
}