
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.tuple.Fields;

/**
 * This topology reads a file and counts the words in that file
 */
public class TopWordFinderTopologyPartB {

  public static void main(String[] args) throws Exception {


    TopologyBuilder builder = new TopologyBuilder();

    Config config = new Config();
    config.setDebug(true);


    /*
    ----------------------TODO-----------------------
    Task: wire up the topology

    NOTE:make sure when connecting components together, using the functions setBolt(name,…) and setSpout(name,…),
    you use the following names for each component:
    FileReaderSpout -> "spout"
    SplitSentenceBolt -> "split"
    WordCountBolt -> "count"
    ------------------------------------------------- */

    String spoutId = "spout";
    String splitId = "split";
    String countId = "count";

    builder.setSpout(spoutId, new FileReaderSpout(args[0]), 5);
    builder.setBolt(splitId, new SplitSentenceBolt(), 8).shuffleGrouping(spoutId);
    builder.setBolt(countId, new WordCountBolt(), 12).fieldsGrouping(splitId, new Fields("word"));

    config.setMaxTaskParallelism(3);

    LocalCluster cluster = new LocalCluster();
    cluster.submitTopology("word-count", config, builder.createTopology());

    //wait for 2 minutes and then kill the job
    Thread.sleep( 2 * 60 * 1000);

    cluster.shutdown();
  }
}

@@ -260,16 +260,18 @@
 artB %7B%0A%0A
+  
   public
@@ -324,16 +324,20 @@
 ion %7B%0A%0A%0A
+    
     Topo
@@ -386,16 +386,20 @@
 );%0A%0A    
+ 
+   
 Config c
@@ -416,24 +416,28 @@
 w Config();%0A
+    
     config.s
@@ -863,16 +863,20 @@
 --- */%0A%0A
+    
     Stri
@@ -897,24 +897,28 @@
 spout%22;%0A    
+   
+ 
 String split
@@ -931,24 +931,28 @@
 split%22;%0A    
+  
+  
 String count
@@ -966,24 +966,28 @@
 ount%22;%0A%0A    
+ 
+   
 builder.setS
@@ -1034,16 +1034,20 @@
 %5D), 5);%0A
+    
     buil
@@ -1121,16 +1121,20 @@
 outId);%0A
+    
     buil
@@ -1225,24 +1225,28 @@
 rd%22));%0A%0A    
+  
+  
 config.setMa
@@ -1263,24 +1263,28 @@
 lelism(3);%0A%0A
+    
     LocalClu
@@ -1318,24 +1318,28 @@
 ster();%0A    
+ 
+   
 cluster.subm
@@ -1403,16 +1403,20 @@
 );%0A%0A    
+   
+ 
 //wait f
@@ -1450,16 +1450,20 @@
 the job%0A
+    
     Thre
@@ -1471,17 +1471,16 @@
 d.sleep(
- 
 2 * 60 *
@@ -1484,24 +1484,28 @@
 0 * 1000);%0A%0A
+    
     cluster.
@@ -1516,14 +1516,16 @@
 down();%0A
+  
   %7D%0A%7D%0A

format file