Learning Storm

<dependencies>
  <dependency>
    <groupId>junit</groupId>
    <artifactId>junit</artifactId>
    <version>3.8.1</version>
    <scope>test</scope>
  </dependency>
  <dependency>
    <groupId>storm</groupId>
    <artifactId>storm</artifactId>
    <version>0.9.0.1</version>
    <scope>provided</scope>
  </dependency>
</dependencies>

<repositories>
  <repository>
    <id>clojars.org</id>
    <url>http://clojars.org/repo</url>
  </repository>
</repositories>

<build>
  <plugins>
    <plugin>
      <artifactId>maven-assembly-plugin</artifactId>
      <version>2.2.1</version>
      <configuration>
        <descriptorRefs>
          <descriptorRef>jar-with-dependencies</descriptorRef>
        </descriptorRefs>
        <archive>
          <manifest>
            <mainClass />
          </manifest>
        </archive>
      </configuration>
      <executions>
        <execution>
          <id>make-assembly</id>
          <phase>package</phase>
          <goals>
            <goal>single</goal>
          </goals>
        </execution>
      </executions>
    </plugin>
  </plugins>
</build>

public class LearningStormSpout extends BaseRichSpout{
  private static final long serialVersionUID = 1L;
  private SpoutOutputCollectorspoutOutputCollector;
  private static final Map<Integer, String> map = new HashMap<Integer, String>();
  static {
    map.put(0, "google");
    map.put(1, "facebook");
    map.put(2, "twitter");
    map.put(3, "youtube");
    map.put(4, "linkedin");
  }
  public void open(Map conf, TopologyContext context, SpoutOutputCollector spoutOutputCollector) {
    // Open the spout
    this.spoutOutputCollector = spoutOutputCollector;
  }

  public void nextTuple() {
    // Storm cluster repeatedly calls this method to emit a continuous
    // stream of tuples.
    final Random rand = new Random();
    // generate the random number from 0 to 4.
    int randomNumber = rand.nextInt(5);
    spoutOutputCollector.emit(new Values(map.get(randomNumber)));

  }

  public void declareOutputFields(OutputFieldsDeclarer declarer) {
    // emit the tuple with field "site"
    declarer.declare(new Fields("site"));
  }
}

public class LearningStormBolt extends BaseBasicBolt{

  private static final long serialVersionUID = 1L;

  public void execute(Tuple input, BasicOutputCollector collector) {
    // fetched the field "site" from input tuple.
    String test = input.getStringByField("site");
    // print the value of field "site" on console.
    System.out.println("Name of input site is : " + test);
  }

  public void declareOutputFields(OutputFieldsDeclarer declarer) {

  }
}

public class LearningStormTopology {
  public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException {
    // create an instance of TopologyBuilder class
    TopologyBuilder builder = new TopologyBuilder();
    // set the spout class
    builder.setSpout("LearningStormSpout", new LearningStormSpout(), 2);
    // set the bolt class
    builder.setBolt("LearningStormBolt", new LearningStormBolt(), 4).shuffleGrouping("LearningStormSpout");

    Config conf = new Config();
    conf.setDebug(true);
    // create an instance of LocalCluster class for 
    // executing topology in local mode.
    LocalCluster cluster = new LocalCluster();

    // LearningStormTopolgy is the name of submitted topology.
    cluster.submitTopology("LearningStormToplogy", conf, builder.createTopology());
    try {
      Thread.sleep(10000);
    } catch (Exception exception) {
      System.out.println("Thread interrupted exception : " + exception);
    }
    // kill the LearningStormTopology
    cluster.killTopology("LearningStormToplogy");
    // shutdown the storm test cluster
    cluster.shutdown();

  }
}

mvn compile exec:java -Dexec.classpathScope=compile 
-
Dexec.mainClass=com.learningstorm.storm_example.LearningStormTopology

cd $ZK_HOME/conf
touch zoo.cfg

tickTime=2000
dataDir=/tmp/zookeeper
clientPort=2181

bin/zkServer.sh start

JMX enabled by default
Using config: /home/root/zookeeper-3.4.5/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED

jps

23074 QuorumPeerMain

bin/zkServer.sh status

JMX enabled by default
Using config: ../conf/zoo.cfg
Mode: standalone

cd $STORM_HOME/conf
vi storm.yaml

storm.zookeeper.servers:
     - "127.0.0.1"
storm.zookeeper.port: 2181
nimbus.host: "127.0.0.1"
storm.local.dir: "/tmp/storm-data"
java.library.path: "/usr/local/lib"
storm.messaging.transport: backtype.storm.messaging.netty.Context
supervisor.slots.ports:
     - 6700
     - 6701
     - 6702
     - 6703

cd $STORM_HOME
bin/storm nimbus

cd $STORM_HOME
bin/storm supervisor

public class LearningStormSingleNodeTopology {
  public static void main(String[] args) {
    TopologyBuilder builder = new TopologyBuilder();
    // set the spout class
    builder.setSpout("LearningStormSpout", new LearningStormSpout(), 4);
    // set the bolt class
    builder.setBolt("LearningStormBolt", new LearningStormBolt(), 2).shuffleGrouping("LearningStormSpout");

    Config conf = new Config();
    conf.setNumWorkers(3);
    try {
      // This statement submit the topology on remote cluster. 
      // args[0] = name of topology
      StormSubmitter.submitTopology(args[0], conf, builder.createTopology());
    }catch(AlreadyAliveException alreadyAliveException) {
      System.out.println(alreadyAliveException);
    } catch (InvalidTopologyException invalidTopologyException) {
      System.out.println(invalidTopologyException);
    }
  }
}

mvn clean install

-----------------------------------------------------------------------
[INFO] ----------------------------------------------------------------
[INFO] BUILD SUCCESS
[INFO] ----------------------------------------------------------------
[INFO] Total time: 58.326s
[INFO] Finished at: Mon Jan 20 00:55:52 IST 2014
[INFO] Final Memory: 14M/116M
[INFO] ----------------------------------------------------------------

bin/storm jar jarName.jar [TopologyMainClass] [Args]

bin/storm jar $PROJECT_HOME/target/storm-example-0.0.1-SNAPSHOT-jar-with-dependencies.jar com.learningstorm.storm_example.LearningStormSingleNodeTopology LearningStormSingleNodeTopology

0    [main] INFO  backtype.storm.StormSubmitter  - Jar not uploaded to master yet. Submitting jar...
7    [main] INFO  backtype.storm.StormSubmitter  - Uploading topology jar /home/root/storm-example/target/storm-example-0.0.1-SNAPSHOT-jar-with-dependencies.jar to assigned location: /tmp/storm-data/nimbus/inbox/stormjar-dfce742b-ca0b-4121-bcbe-1856dc1846a4.jar
19   [main] INFO  backtype.storm.StormSubmitter  - Successfully uploaded topology jar to assigned location: /tmp/storm-data/nimbus/inbox/stormjar-dfce742b-ca0b-4121-bcbe-1856dc1846a4.jar
19   [main] INFO  backtype.storm.StormSubmitter  - Submitting topology LearningStormSingleNodeTopologyin distributed mode with conf{"topology.workers":3}
219  [main] INFO  backtype.storm.StormSubmitter  - Finished submitting topology: LearningStormSingleNodeTopology

jps

26827 worker
26530 supervisor
26824 worker
26468 nimbus
15987 QuorumPeerMain
26822 worker

bin/storm deactivate topologyName

bin/storm deactivate LearningStormSingleNodeTopology

0    [main] INFO  backtype.storm.thrift  - Connecting to Nimbus at localhost:6627r
76   [main] INFO  backtype.storm.command.deactivate  - Deactivated topology: LearningStormSingleNodeTopology

bin/storm activate topologyName

bin/storm activate LearningStormSingleNodeTopology

0    [main] INFO  backtype.storm.thrift  - Connecting to Nimbus at localhost:6627
65   [main] INFO  backtype.storm.command.activate  - Activated topology: LearningStormSingleNodeTopology

bin/storm kill LearningStormSingleNodeTopology

0    [main] INFO  backtype.storm.thrift  - Connecting to Nimbus at localhost:6627
80   [main] INFO  backtype.storm.command.kill-topology  - Killed topology: LearningStormSingleNodeTopology

jps

26530 supervisor
27193 Jps
26468 nimbus
15987 QuorumPeerMain