Necesito ayuda para implementar este algoritmo con el mapa Hadoop MapReduce

Question

tengo un algoritmo que pasará por un gran conjunto de datos leerá algunos archivos de texto y buscará términos específicos en esas líneas. Lo tengo implementado en Java, pero no quise publicar el código para que no parezca que estoy buscando alguien que lo implemente, pero es cierto, realmente necesito mucha ayuda. Esto no fue planeado para mi proyecto, pero el conjunto de datos resultó ser enorme, así que la maestra me dijo que tenía que hacerlo así.

EDITAR (no me aclaro i Previos versión) El conjunto de datos que tengo es en un clúster Hadoop, y yo debería hacer que su implementación MapReduce

que estaba leyendo acerca de MapReduce y thaught que primero haga el implementación estándar y luego será más o menos fácil hacerlo con mapreduce. Pero no sucedió, ya que el algoritmo es bastante estúpido y nada especial, y el mapa se reduce ... no puedo entenderlo.

Así que aquí es poco seudo código de mi algoritmo

LIST termList (there is method that creates this list from lucene index) 
FOLDER topFolder 

INPUT topFolder 
IF it is folder and not empty 
    list files (there are 30 sub folders inside) 
    FOR EACH sub folder 
     GET file "CheckedFile.txt" 
     analyze(CheckedFile) 
    ENDFOR 
END IF 


Method ANALYZE(CheckedFile) 

read CheckedFile 
WHILE CheckedFile has next line 
    GET line 
    FOR(loops through termList) 
      GET third word from line 
      IF third word = term from list 
     append whole line to string buffer 
    ENDIF 
ENDFOR 
END WHILE 
OUTPUT string buffer to file 

Además, como se puede ver, cada vez que "analizar" se llama, nuevo archivo tiene que ser creado, entendí que MapReduce es difícil para escribir en muchas salidas ???

Entiendo mapreduce la intuición, y mi ejemplo parece perfectamente adecuado para mapreduce, pero cuando se trata de hacer esto, obviamente no sé lo suficiente y ¡estoy PEGADO!

Por favor ayuda.

Answer 1

Puede usar un reductor vacío y dividir su trabajo para ejecutar un solo asignador por archivo. Cada mapeador creará su propio archivo de salida en su carpeta de salida.

Answer 2

Map Reduce se implementa fácilmente utilizando algunas agradables funciones de simultaneidad Java 6, especialmente Future, Callable y ExecutorService.

he creado un rescatable que analizará un archivo en la forma en que ha especificado

public class FileAnalyser implements Callable<String> { 

    private Scanner scanner; 
    private List<String> termList; 

    public FileAnalyser(String filename, List<String> termList) throws FileNotFoundException { 
    this.termList = termList; 
    scanner = new Scanner(new File(filename)); 
    } 

    @Override 
    public String call() throws Exception { 
    StringBuilder buffer = new StringBuilder(); 
    while (scanner.hasNextLine()) { 
     String line = scanner.nextLine(); 
     String[] tokens = line.split(" "); 
     if ((tokens.length >= 3) && (inTermList(tokens[2]))) 
     buffer.append(line); 
    } 
    return buffer.toString(); 
    } 

    private boolean inTermList(String term) { 
    return termList.contains(term); 
    } 
} 

Necesitamos crear un nuevo exigible para cada archivo encontrado y la enviará al servicio de ejecutor. El resultado de la presentación es un futuro que podemos usar más adelante para obtener el resultado del análisis del archivo.

public class Analayser { 

    private static final int THREAD_COUNT = 10; 

    public static void main(String[] args) { 

    //All callables will be submitted to this executor service 
    //Play around with THREAD_COUNT for optimum performance 
    ExecutorService executor = Executors.newFixedThreadPool(THREAD_COUNT); 

    //Store all futures in this list so we can refer to them easily 
    List<Future<String>> futureList = new ArrayList<Future<String>>(); 

    //Some random term list, I don't know what you're using. 
    List<String> termList = new ArrayList<String>(); 
    termList.add("terma"); 
    termList.add("termb"); 

    //For each file you find, create a new FileAnalyser callable and submit 
    //this to the executor service. Add the future to the list 
    //so we can check back on the result later 
    for each filename in all files { 
     try { 
     Callable<String> worker = new FileAnalyser(filename, termList); 
     Future<String> future = executor.submit(worker); 
     futureList.add(future); 
     } 
     catch (FileNotFoundException fnfe) { 
     //If the file doesn't exist at this point we can probably ignore, 
     //but I'll leave that for you to decide. 
     System.err.println("Unable to create future for " + filename); 
     fnfe.printStackTrace(System.err); 
     } 
    } 

    //You may want to wait at this point, until all threads have finished 
    //You could maybe loop through each future until allDone() holds true 
    //for each of them. 

    //Loop over all finished futures and do something with the result 
    //from each 
    for (Future<String> current : futureList) { 
     String result = current.get(); 
     //Do something with the result from this future 
    } 
    } 
} 

Mi ejemplo aquí está lejos de ser completo y está lejos de ser eficiente. No he considerado el tamaño de la muestra, si es realmente enorme se podía mantener un bucle sobre el futureList, la eliminación de los elementos que han terminado, algo similar a:

while (futureList.size() > 0) { 
     for (Future<String> current : futureList) { 
     if (current.isDone()) { 
      String result = current.get(); 
      //Do something with result 
      futureList.remove(current); 
      break; //We have modified the list during iteration, best break out of for-loop 
     } 
     } 
} 

Como alternativa puede implementar una configuración del tipo de productor-consumidor, donde la el productor envía callables al servicio del ejecutor y produce un futuro y el consumidor toma el resultado del futuro y descarta el futuro.

Esto podría requerir que el producto y el consumidor sean hilos, y una lista sincronizada para agregar/eliminar futuros.

Cualquier pregunta, pregunte.