Trouver des lignes correspondantes avec Hadoop/MapReduce

Question

Je joue avec Hadoop et j'ai mis en place un cluster à deux nœuds sur Ubuntu. L'exemple WordCount fonctionne très bien.

Maintenant, je voudrais écrire mon propre programme MapReduce pour analyser des données de log (principale raison: il semble simple et j'ai beaucoup de données)

Chaque ligne du hast journal ce format

<UUID> <Event> <Timestamp> 

où l'événement peut être INIT, START, STOP, ERROR et un autre. Ce qui m'intéresse le plus, c'est le temps écoulé entre les événements START et STOP pour le même UUID.

Par exemple, mon journal contient des entrées comme celles-ci

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

Mon actuelle, le programme linéaire lit les fichiers, se souvient des événements de départ en mémoire, et écrit le temps écoulé à un fichier une fois qu'il a trouvé la événement d'extrémité correspondante (lignes avec d'autres événements sont actuellement ignorés, les événements d'erreur invalident un UUID et il sera ignoré, aussi)

Je voudrais au port ce à un programme Hadoop/MapReduce. Mais je ne suis pas sûr de savoir comment faire la correspondance des entrées. La division/Tokenizing du fichier est facile, et je suppose que trouver les correspondances sera une classe de réduction. Mais à quoi cela ressemblerait-il? Comment trouver des entrées dans un travail MapReduce?

Veuillez garder à l'esprit que mon objectif principal est de comprendre Hadopo/MapReduce; Les liens vers Pig et d'autres programmes Apache sont les bienvenus, mais je voudrais résoudre celui-ci avec de purs Hadoop/MapReduce. Je vous remercie.

¹⁾Étant donné que le journal est pris d'une application en cours d'exécution, certains événements de départ peuvent pas encore des événements de fin correspondant et il y aura finaux événements sans startevents, en raison de logfile division

Answer 1

Si vous émettez l'UUID sur la carte comme la clé: emit(<uuid>, <event, timestamp>) vous recevrez dans votre réduire tous les événements de cette UUID: key = UUID, values = {<event1, timestamp1>, <event2, timestamp2>}

Ensuite, vous pouvez trier les événements sur horodatage et décider de les émettre dans un fichier résultant ou non.

Bonus: vous pouvez utiliser job.setSortComparatorClass(); pour définir votre propre classe de tri, de sorte que vous aurez vos entrées déjà triées sur leurs horodatages en réduire:

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

Answer 2

Je pense vous pouvez le faire en faisant en sorte que votre fonction de carte produise l'UUID comme clé et le reste de la ligne comme valeur. Alors la fonction de réduction passera une collection de toutes les entrées de journal avec le même UUID. En les traitant, il peut suivre les différents événements qu'il voit et agir en conséquence. Par exemple, lorsqu'il voit un événement START, il peut définir une variable locale sur l'heure extraite de la ligne de départ, puis quand il voit un STOP événement, il peut en extraire l'heure, soustraire l'heure de départ, et sortir la différence (et faire de même si elle voit le STOP avant le START).