java - tutorial - Encontrar líneas coincidentes con Hadoop/MapReduce

mapreduce mongodb

Question

Estoy jugando con Hadoop y he configurado un clúster de dos nodos en Ubuntu. El ejemplo de WordCount funciona bien.

Ahora me gustaría escribir mi propio programa MapReduce para analizar algunos datos de registro (razón principal: parece simple y tengo muchos datos)

Cada línea en el registro tiene este formato

<UUID> <Event> <Timestamp>

donde el evento puede ser INIT, START, STOP, ERROR y algunos otros. Lo que más me interesa es el tiempo transcurrido entre los eventos START y STOP para el mismo UUID.

Por ejemplo, mi registro contiene entradas como estas

35FAA840-1299-11DF-8A39-0800200C9A66 START 1265403584
[...many other lines...]
35FAA840-1299-11DF-8A39-0800200C9A66 STOP 1265403777

Mi programa lineal actual lee los archivos, recuerda los eventos de inicio en memoria y escribe el tiempo transcurrido en un archivo una vez que encontró el evento final correspondiente (las líneas con otros eventos se ignoran actualmente, los eventos ERROR invalidan un UUID y lo hará ser ignorado, también) ¹

Me gustaría transferir esto a un programa Hadoop / MapReduce. Pero no estoy seguro de cómo hacer la coincidencia de las entradas. Dividir / Tokenizar el archivo es fácil, y supongo que encontrar las coincidencias será una Clase Reducida. ¿Pero cómo se vería eso? ¿Cómo encuentro entradas mathing en un trabajo de MapReduce?

Tenga en cuenta que mi objetivo principal es comprender Hadopo / MapReduce; enlaces a Pig y otros programas de Apache son bienvenidos, pero me gustaría resolver este con Hadoop / MapReduce puro. Gracias.

¹⁾ Dado que el registro se toma de una aplicación en ejecución, es posible que algunos eventos de inicio aún no tengan los eventos finales correspondientes y habrá eventos finales sin eventos de secuencia, debido a la división del archivo de registro.

Answer 1

Creo que podría hacer esto haciendo que su función de mapa emita el UUID como su clave y el resto de la línea como su valor. A continuación, la función de reducción recibirá una colección de todas las entradas de registro con el mismo UUID. A medida que los procesa, puede hacer un seguimiento de los diversos eventos que ve y tomar medidas en consecuencia; por ejemplo, cuando ve un evento START puede establecer una variable local a la hora extraída de la línea de inicio, y luego cuando ve una PARADA el evento puede extraer el tiempo de él, restar la hora de inicio y generar la diferencia (y hacer lo mismo si ve la PARADA antes del COMIENZO).

Answer 2

Si usted emite el UUID en el mapa como clave: emit(<uuid>, <event, timestamp>) recibirá en su reducir todos los eventos de este UUID: key = UUID, values = {<event1, timestamp1>, <event2, timestamp2>}

Luego puede ordenar los eventos en timestamp y decidir si los emitirá en un archivo resultante o no.

Bonificación: puede usar job.setSortComparatorClass(); para configurar su propia clase de clasificación, por lo que obtendrá sus entradas ya ordenadas en sus marcas de tiempo en reducir:

public static class BNLSortComparator extends Text.Comparator {
  public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
    String sb1, sb2;
    try {
      sb1 = Text.decode(b1, s1, l1);
      ...