我正在使用这两个实时数据流框架处理器。我已经到处搜索,但是我找不到这两个框架之间的大差异。特别是我想知道它们如何基于数据大小或拓扑结构进行工作。
我正在使用这两个实时数据流框架处理器。我已经到处搜索,但是我找不到这两个框架之间的大差异。特别是我想知道它们如何基于数据大小或拓扑结构进行工作。
TridentState wordCounts =
topology.newStream("spout1", spout)
.each(new Fields("sentence"), new Split(), new Fields("word"))
.groupBy(new Fields("word"))
.persistentAggregate(new MemoryMapState.Factory(), new Count(), new Fields("count"))
.parallelismHint(6);
DataStream<WordWithCount> windowCounts = text
.flatMap(new FlatMapFunction<String, WordWithCount>() {
@Override
public void flatMap(String value, Collector<WordWithCount> out) {
for (String word : value.split("\\s")) {
out.collect(new WordWithCount(word, 1L));
}
}
})
.keyBy("word")
.timeWindow(Time.seconds(5), Time.seconds(1))
.reduce(new ReduceFunction<WordWithCount>() {
@Override
public WordWithCount reduce(WordWithCount a, WordWithCount b) {
return new WordWithCount(a.word, a.count + b.count);
}
});