(Twitter) Storm's Window On Aggregation

Question

Sto giocando con Storm, e mi chiedo dove Storm specifichi (se possibile) le dimensioni della finestra (ruzzolando/scorrendo) su un'aggregazione. Per esempio. Se vogliamo trovare gli argomenti di tendenza per l'ora precedente su Twitter. Come specifichiamo che un bullone dovrebbe restituire i risultati per ogni ora? Questo è fatto programmaticamente all'interno di ogni bullone? O è un modo per specificare una "finestra"?

Answer 1

Disclaimer: Ho scritto gli Argomenti di tendenza con l'articolo Tempesta a cui si fa riferimento gakhov in his answer above.

Direi che la pratica migliore è utilizzare il cosiddetto tick tuples in Storm 0.8+. Con questi è possibile configurare i propri beccucci/bulloni per essere avvisati a determinati intervalli di tempo (ad esempio, ogni dieci secondi o ogni minuto).

Ecco un semplice esempio che configura il componente in questione per ricevere tuple tick ogni dieci secondi:

// in your spout/bolt 
@Override 
public Map<String, Object> getComponentConfiguration() { 
    Config conf = new Config(); 
    int tickFrequencyInSeconds = 10; 
    conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, tickFrequencyInSeconds); 
    return conf; 
} 

È quindi possibile utilizzare un interruttore condizionale nel metodo tua bocca/del bullone execute() di distinguere "normali" tuple in arrivo dalle speciali tuple di tic tac. Per esempio:

// in your spout/bolt 
@Override 
public void execute(Tuple tuple) { 
    if (isTickTuple(tuple)) { 
     // now you can trigger e.g. a periodic activity 
    } 
    else { 
     // do something with the normal tuple 
    } 
} 

private static boolean isTickTuple(Tuple tuple) { 
    return tuple.getSourceComponent().equals(Constants.SYSTEM_COMPONENT_ID) 
     && tuple.getSourceStreamId().equals(Constants.SYSTEM_TICK_STREAM_ID); 
} 

Anche in questo caso, ho scritto una bella detailed blog post a fare questo a Storm pochi giorni fa, come ha sottolineato gakhov (senza vergogna!).

Answer 2

Aggiungere un nuovo beccuccio con grado di parallelismo pari a 1 e farlo emettere un segnale vuoto, quindi Utils.sleep fino alla volta successiva (operazione completata in nextTuple). Quindi, collega tutti i bulloni pertinenti a quel beccuccio usando tutti i raggruppamenti, così tutte le loro istanze riceveranno lo stesso segnale.