Come contare le parole uniche in un flusso?

Question

C'è un modo per contare il numero di parole uniche in un flusso con Flink Streaming? I risultati sarebbero un flusso di numeri che continua ad aumentare.

Answer 1

È possibile risolvere il problema memorizzando tutte le parole che hai già visto. Con questa conoscenza puoi filtrare tutte le parole duplicate. Il resto può quindi essere conteggiato da un operatore di mappa con parallelismo 1. Il seguente frammento di codice fa esattamente questo.

val env = StreamExecutionEnvironment.getExecutionEnvironment 

val inputStream = env.fromElements("foo", "bar", "foobar", "bar", "barfoo", "foobar", "foo", "fo") 

// filter words out which we have already seen 
val uniqueWords = inputStream.keyBy(x => x).filterWithState{ 
    (word, seenWordsState: Option[Set[String]]) => seenWordsState match { 
    case None => (true, Some(HashSet(word))) 
    case Some(seenWords) => (!seenWords.contains(word), Some(seenWords + word)) 
    } 
} 

// count the number of incoming (first seen) words 
val numberUniqueWords = uniqueWords.keyBy(x => 0).mapWithState{ 
    (word, counterState: Option[Int]) => 
    counterState match { 
     case None => (1, Some(1)) 
     case Some(counter) => (counter + 1, Some(counter + 1)) 
    } 
}.setParallelism(1) 

numberUniqueWords.print(); 

env.execute()