### Apache Storm WordCount 示例代码 Apache Storm 是一种分布式实时计算系统，能够处理大规模的数据流。WordCount 是常见的入门级案例之一，用于统计文本中的单词频率。 以下是基于 Java 编写的 Apache Storm 的 WordCount 示例代码： ```java import backtype.storm.Config; import backtype.storm.LocalCluster; import backtype.storm.topology.BasicOutputCollector; import backtype.storm.topology.OutputFieldsDeclarer; import backtype.storm.topology.TopologyBuilder; import backtype.storm.tuple.Fields; import backtype.storm.tuple.Tuple; import backtype.storm.tuple.Values; public class WordCountTopology { public static class SentenceSpout extends BaseRichSpout { private SpoutOutputCollector collector; private String[] sentences = {"the cow jumped over the moon", "an apple a day keeps the doctor away", "four score and seven years ago", "snow white and the seven dwarfs"}; private int index = 0; @Override public void open(Map config, TopologyContext context, SpoutOutputCollector collector) { this.collector = collector; } @Override public void nextTuple() { String sentence = sentences[index]; index = (index + 1) % sentences.length; collector.emit(new Values(sentence)); } @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("sentence")); } } public static class SplitSentenceBolt implements IRichBolt { private OutputCollector collector; @Override public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) { this.collector = collector; } @Override public void execute(Tuple input) { String sentence = input.getString(0); for (String word : sentence.split("\\s+")) { collector.emit(input, new Values(word.toLowerCase())); } collector.ack(input); } @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("word")); } @Override public Map<String, Object> getComponentConfiguration() { return null; } } public static class CountWordsBolt implements IRichBolt { private HashMap<String, Integer> counts = new HashMap<>(); private OutputCollector collector; @Override public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) { this.collector = collector; } @Override public void execute(Tuple input) { String word = input.getString(0); Integer count = counts.getOrDefault(word, 0); counts.put(word, count + 1); System.out.println(word + ": " + counts.get(word)); collector.ack(input); } @Override public void declareOutputFields(OutputFieldsDeclarer declarer) {} @Override public Map<String, Object> getComponentConfiguration() { return null; } } public static void main(String[] args) throws Exception { TopologyBuilder builder = new TopologyBuilder(); builder.setSpout("spout", new SentenceSpout(), 5); builder.setBolt("split", new SplitSentenceBolt(), 8).shuffleGrouping("spout"); builder.setBolt("count", new CountWordsBolt(), 12).fieldsGrouping("split", new Fields("word")); Config config = new Config(); config.setDebug(true); LocalCluster cluster = new LocalCluster(); cluster.submitTopology("word-count-topology", config, builder.createTopology()); Thread.sleep(10000); cluster.shutdown(); } } ``` #### 解析 - **SentenceSpout**: 这是一个简单的 spout，负责生成句子[^1]。 - **SplitSentenceBolt**: 将输入的句子拆分为单个单词，并将其发送到下一个组件[^4]。 - **CountWordsBolt**: 统计每个单词出现的次数，并打印结果[^3]。 此拓扑结构通过 `shuffleGrouping` 和 `fieldsGrouping` 实现了数据分布和聚合操作[^5]。 --- ###