Trovare le linee corrispondenti con Hadoop / MapReduce

Question

sto giocando intorno con Hadoop e hanno creato un cluster a due nodi su Ubuntu. L'esempio WordCount funziona bene.

Ora mi piacerebbe scrivere il mio programma di MapReduce di analizzare alcuni dati di log (motivo principale: sembra semplice e ho un sacco di dati)

Ogni riga del registro hai questo formato

<UUID> <Event> <Timestamp>

dove evento può essere INIT, START, STOP, ERROR e qualche altro. Ciò che mi interessa di più è il tempo trascorso tra Start e Stop eventi per lo stesso UUID.

Per esempio, il mio registro contiene voci come questi

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

Il mio attuale programma, lineare legge attraverso i file, ricorda gli eventi di avvio in memoria, e scrive il tempo trascorso in un file, una volta trovato l'evento di fine corrispondente (linee con altri eventi sono attualmente ignorati, gli eventi di errore invalidano un UUID e sarà ignorato, anche) ¹

Vorrei porta questo ad un programma / MapReduce Hadoop. Ma io non sono sicuro di come fare l'abbinamento delle voci. Splitting / creazione di token il file è semplice, e credo che trovare le partite sarà un Ridurre-Class. Ma come vorrei che assomiglia? Come faccio a trovare mathing voci in un MapReduce lavoro?

Si prega di tenere presente che il mio obiettivo principale è quello di comprendere Hadopo / MapReduce; link ad maiale e altri programmi Apache sono i benvenuti, ma mi piacerebbe risolvere questo uno con puro Hadoop / MapReduce. Grazie.

¹⁾ Dal momento che il registro è tratto da un'applicazione in esecuzione, alcuni eventi di avvio potrebbero non hanno ancora corrispondenti eventi finali e ci saranno finali eventi senza startevents, a causa di file di log splitting

Answer 1

Se si emettono l'UUID in mappa come chiave: emit(<uuid>, <event, timestamp>) riceverai nella tua ridurre tutti gli eventi di questo UUID: key = UUID, values = {<event1, timestamp1>, <event2, timestamp2>}

Quindi è possibile ordinare gli eventi in timestamp e decidere se li emettono in un file risultante oppure no.

Bonus: è possibile utilizzare job.setSortComparatorClass(); per impostare la propria classe di ordinamento, così avrai le voci già ordinato il loro timestamp a ridurre:

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

Penso che si potrebbe fare questo facendo l'output mappa funzione l'UUID come chiave e il resto della linea come il suo valore. Poi la funzione di ridurre verrà passato una raccolta di tutte le voci di registro con lo stesso UUID. Come li elabora può tenere traccia dei vari eventi che vede e agire di conseguenza - per esempio, quando si vede un evento di avvio è possibile impostare una variabile locale al tempo estratto dalla linea di partenza, e poi quando si vede uno STOP evento si può estrarre il tempo da esso, sottrarre l'ora di inizio, e l'uscita la differenza (e fare allo stesso modo se si vede l'ARRESTO prima della partenza).