Compilazione ed esecuzione dinamica di un processo Hadoop da un altro file Java

Sto cercando di scrivere un file Java che riceve il codice sorgente di un lavoro MapReduce, lo compila dinamicamente ed esegue il lavoro su un cluster Hadoop. Per raggiungere questo, ho scritto 3 metodi chiamati compile (), makeJAR () e run_Hadoop_Job (). Tutto funziona bene con la compilazione e la creazione del file JAR. Tuttavia, quando il lavoro viene inviato a Hadoop, non appena il lavoro inizia, affronta problemi con la ricerca delle classi Mapper/Reducer richieste e lancia una ClassNotFoundException per entrambi il Mapper_Class e Reducer_Class *(java.lang.ClassNotFoundException: reza.Cloud.Mapper_Reducer_Classes Ma Mapper_Class.classe)*. So che dovrebbe esserci qualcosa di sbagliato nel modo in cui ho fatto riferimento alle classi Mapper/Reducer richieste, ma non sono stato in grado di capirlo dopo diversi. Qualsiasi aiuto / suggerimento su come risolvere il problema è molto apprezzato.

Per quanto riguarda i dettagli del progetto: Ho un file chiamato "rCloud_test/src/reza/Mapper_Reducer_Classes.Java" che contiene il codice sorgente per Mapper_Class e Reducer_Class. Questo file viene infine ricevuto durante il runtime, ma per ora ho copiato l'esempio di Hadoop WordCount e lo memorizzo localmente nella stessa cartella del mio file di classe principale: rCloud_test/src/reza / Platform2.Java.

Qui sotto puoi vedere il metodo main() della Platform2.java che è la mia classe principale per questo progetto:

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

Il metodo che mi sta causando tutti i problemi è run_Hadoop_Job () che è come sotto:

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

Se necessario, ecco il codice sorgente per il metodo 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;
    }   
}

E il codice sorgente per il metodo 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();
  }
}

E infine i parametri fissi utilizzati per accedere ai file:

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

Come risultato dell'esecuzione di Platform2, la struttura del progetto su disco appare come di seguito:

RCloud_test/classes/reza/rCloud/Platform2.classe: contiene la classe Platform2 rCloud_test/dyn_classes/reza/ rCloud / contiene le classi per Mapper_Reducer_Classes.classe, Mapper_Reducer_Classes Ma Mapper_Class.classe e Mapper_Reducer_Classes Reduc Reducer_Class.classe rCloud_test / dyn_jar / jar_file.jar contiene il file jar creato

REVSED: ecco il codice sorgente per le classi 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));
    }
}
}