在 Flink Streaming 中,有没有一种方法可以统计数据流中唯一单词的数量?结果应该是一个数字流,不断增加。
在 Flink Streaming 中,有没有一种方法可以统计数据流中唯一单词的数量?结果应该是一个数字流,不断增加。
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()
filterWithState
中的字符串集变得太大,它会导致OOM或性能下降吗? - MaximRocksDBStateBackend
就是这样一种状态后端。如果您使用内存状态后端,则必须定期清除状态,否则可能会遇到 OOM 问题。 - Till RohrmannRocksDBStateBackend
后端的情况下,保存/恢复操作的复杂度是O(N),其中N是集合中元素的数量,即此后端是否始终保存/恢复Set的所有元素或仅更改的元素? - MaximValueState
抽象,该抽象始终保存/恢复完整的Set
。但是,可能还可以使用ListState
抽象使检查点增量化。 - Till RohrmannfilterWithState
只在1.1版本中可用吗?我在Flink 1.0.0中找不到它。 - JunfilterWithState
方法自 Flink 1.0.0 版本起成为 Scala API 的一部分。然而,它仅适用于KeyedStreams
。这意味着您必须先在流上调用keyBy
方法。 - Till Rohrmann