对于每条推文：
对于tweet中的每个单词：
试着把每个单词都加到一个集合中。
如果单词已存在于集合中，请将其从集合中删除。
计算包含该tweet的单词集的大小。

“问题”是“获取所有tweet上不同单词的计数”和strom作为流处理器之间的不匹配。您要回答的查询只能在有限的tweet集上计算。然而，在流处理中，您处理一个潜在的无限输入数据流。
如果您有一组有限的tweet，则可能需要使用批处理框架，如flink、spark或mapreduce。如果你确实有无限多的推特，你必须重新措辞你的问题。。。
正如您已经提到的，您实际上想要“循环所有tweets”。当你这么做流处理，没有这样的概念。您有无限多个输入元组，而storm适用于 execute() 在每一种情况下（也就是说，你可以把它想象成storm“在输入上自动循环”——即使“循环”不是正确的术语）。由于您的计算是“覆盖所有tweet”，因此您需要在bolt代码中维护一个状态，这样您就可以为每个tweet更新这个状态。storm中状态的simples形式将是bolt类中的成员变量。

public class MyBolt implements ??? {
    // this is your "state" variable
    private final Set<String> allWords = new HashSet<String>();

    public void execute(TridentTuple tuple, TridentCollector collector) {
        Tweet tweet = (Tweet)tuple.getValue(0);        
        String tweetBody = tweet.getBody();
        String words[] = tweetBody.toLowerCase().split(regex);
        for(String w : words) {
            // as allWords is a set, you cannot add the same word twice
            // the second "add" call on the same word will just be ignored
           // thus, allWords will contain each word exactly once
            this.allWords.add(w);
        }
    }
}

现在，这个代码没有发出任何东西，因为不清楚您实际想要发出什么？在流处理中，没有结束，您不能说“emit the final count of words，contained in allWords ". 你能做的是，它在每次更新后发出当前计数。。。为此，添加 collector.emit(new Values(this.allWords.size())); 结束时 execute() .
此外，我想补充的是，所提出的解决方案只有在没有应用并行性的情况下才能正常工作 MyBolt --否则，示例上的不同集合可能包含相同的单词。要解决这个问题，需要将每条tweet标记为无状态的bolt中的单词，并将这些单词流放在一个已采用的tweet中 MyBolt 使用内部 Set 作为国家。的输入数据 MyBolt 还必须通过接收数据 fieldsGrouping 以确保每个示例上有不同的词集。