- Real-time Analytics with Storm and Cassandra
- Shilpi Saxena
- 412字
- 2021-07-23 19:52:09
Executing the topology from Command Prompt
Once the UI is visible and all the daemons are started, the topology can be submitted on Nimbus using the following command:
storm jar storm-starter-0.0.1-SNAPSHOT-jar-with-dependencies.jar storm.starter.WordCountTopology WordCount -c nimbus.host=localhost
The Storm UI with the WordCount
topology running in distributed mode is shown here. It depicts the topology state, uptime, and other details (we shall discuss the features of the UI in detail in a later chapter). We can kill the topology from the UI.

Tweaking the WordCount topology to customize it
Now that we have deployed the WordCount
topology in distributed mode, let's tweak the code in the bolts a bit to write WordCount
onto a file. To achieve this, we will proceed with the following steps:
- We intend to create a new bolt,
FileWriterBolt
, to achieve this. OpenWordCountTopology.java
and add the following snippet toWordCountTopology.java
:public static class FileWriterBolt extends BaseBasicBolt { Map<String, Integer> counts = new HashMap<String, Integer>(); @Override public void execute(Tuple tuple, BasicOutputCollector collector) { String word = tuple.getString(0); Integer count = counts.get(word); if(count==null) {count = 0; count = 0; } count++; counts.put(word, count); OutputStream ostream; try { ostream = new FileOutputStream("~/wordCount.txt", true); ostream.write(word.getBytes()); ostream.close(); } catch (IOException e) { // TODO Auto-generated catch block e.printStackTrace(); } collector.emit(new Values(word, count)); } @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("word", "count")); }
- Next we have to make changes to the
main()
method to use this new bolt instead ofWordCount Bolt()
; here is the snippet:// instantiates the new builder object TopologyBuilder builder = new TopologyBuilder(); // Adds a new spout of type "RandomSentenceSpout" with a parallelism hint of 5 builder.setSpout("spout", new RandomSentenceSpout(), 5); //Adds a new bolt to the topology of type "SplitSentence" with parallelism of 8 builder.setBolt("split", new SplitSentence(), 8).shuffleGrouping("spout"); //Adds a new bolt to the topology of type "SplitSentence" with parallelism of 8 //builder.setBolt("count", new FileWriterBolt()(), 12).fieldsGrouping("split", new Fields("word"));
- Next, you can execute the topology using Eclipse, run it as Java, and the output will be saved into a file called
wordCount.txt
in your home directory. - To run in distributed mode, use the following steps:
- Compile the topology changes to generate a new Storm-starter project using the following command line:
mvn clean install
- Copy
storm-starter-0.0.1-SNAPSHOT-jar-with-dependencies.jar
from the target folder under the starter project to Nimbus, let's say, at/home/admin/topology/
. - Submit the topology using the following command:
storm jar /home/admin/topology/storm-starter-0.0.1-SNAPSHOT- jar-with-dependencies.jar storm.starter.WordCountTopology WordCount -c nimbus.host=localhost
- Compile the topology changes to generate a new Storm-starter project using the following command line:
- The output will be the same as the
WordCount
topology executed in the figure in the preceding section.
推薦閱讀
- 從零開始構建企業級RAG系統
- Raspberry Pi for Python Programmers Cookbook(Second Edition)
- Visual C++串口通信開發入門與編程實踐
- Visual C++數字圖像模式識別技術詳解
- Mastering RStudio:Develop,Communicate,and Collaborate with R
- Python面向對象編程:構建游戲和GUI
- Hands-On Full Stack Development with Go
- Python算法指南:程序員經典算法分析與實現
- 響應式架構:消息模式Actor實現與Scala、Akka應用集成
- Java網絡編程實戰
- Frank Kane's Taming Big Data with Apache Spark and Python
- C++20高級編程
- Canvas Cookbook
- 零基礎學Python編程(少兒趣味版)
- Java Hibernate Cookbook