如何计算流中的唯一单词

Question

有没有一种方法可以通过Flink Streaming计算流中唯一单词的数量？结果将是一个不断增加的数字流。

Answer 1

您可以通过存储您已经看到的所有单词来解决问题。有了这些知识，你可以过滤掉所有重复的单词。然后，剩余部分可以由具有并行度1的映射运算符进行计数。下面的代码片段正是这样做的。

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()