Setting up your development environment
1. download j2se 6 SDK from http://www.oracle.com/technetwork/java/javase/downloads/index.html chmod 775 jdk-6u35-linux-64.bin yes | jdk-6u35-linux-64.bin mv jdk1.6.0_35 /opt ln -s /opt/jdk1.6.0_35/bin/java /usr/bin ln -s /opt/jdk1.6.0_35/bin/javac /usr/bin export JAVA_HOME=/opt/jdk1.6.0_35 export PATH=$PATH:$JAVA_HOME/bin 2. install Git sudo apt-get install git 3. install Maven sudo apt-get install mvn 4. install Puppet, Vagrant, and VirtualBox sudo apt-get install virtualbox puppet vagrant 5. install Eclipse sudo apt-get install eclipse
Distributed version control
1. create project directory mkdir FirstGitProject cd FirstGitProject git init 2. create some filess in the repository touch README.txt vim README.txt 3. review the status of the repository git status 4. add all files and folders manually git add README.txt 5. commit the file git commit -a -m "The first commit" 6. add the remote repository to local repository and push the changes git remote add origin https://[user]@bitbucket.org/[user]/firstgitproject.git git push origin master
Creating a "Hello World" topology
1. create a new project folder and init Git repository
mkdir HelloWorld
cd HelloWorld
git init
2. create Maven project file
vi pom.xml
3. create the basic XML tags and project metadata
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>storm.cookbook</groupId>
<artifactId>hello-world</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>hello-world</name>
<url>https://bitbucket.org/[user]/hello-world</url>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
</project>
4. declare the Maven repositories to fetch the dependencies from
<repositories>
<repository>
<id>github-releases</id>
<url>http://oss.sonatype.org/content/repositories/github-releases/</url>
</repository>
<repository>
<id>clojars.org</id>
<url>http://clojars.org/repo</url>
</repository>
<repository>
<id>twitter4j</id>
<url>http://twitter4j.org/maven2</url>
</repository>
</repositories>
5. declare the dependencies
<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.8.1</version>
<!-- keep storm out of the jar-with-dependencies -->
<scope>provided</scope>
</dependency>
<dependency>
<groupId>com.googlecode.json-simple</groupId>
<artifactId>json-simple</artifactId>
<version>1.1</version>
</dependency>
</dependencies>
6. add the build plugin definitions
<build>
<plugins>
<plugin>
<artifactId>maven-assembly-plugin</artifactId>
<configuration>
<descriptorRefs>
<descriptorRef>jar-with-dependencies</descriptorRef>
</descriptorRefs>
<archive>
<manifest>
<mainClass></mainClass>
</manifest>
</archive>
</configuration>
<executions>
<execution>
<id>make-assembly</id>
<phase>package</phase>
<goals>
<goal>single</goal>
</goals>
</execution>
</executions>
</plugin>
<plugin>
<groupId>com.theoryinpractise</groupId>
<artifactId>clojure-maven-plugin</artifactId>
<version>1.3.8</version>
<extensions>true</extensions>
<configuration>
<sourceDirectories>
<sourceDirectory>src/clj</sourceDirectory>
</sourceDirectories>
</configuration>
<executions>
<execution>
<id>compile</id>
<phase>compile</phase>
<goals>
<goal>compile</goal>
</goals>
</execution>
<execution>
<id>test</id>
<phase>test</phase>
<goals>
<goal>test</goal>
</goals>
</execution>
</executions>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<source>1.6</source>
<target>1.6</target>
</configuration>
</plugin>
</plugins>
</build>
7. complete required folder structure
src/main/java
src/test
8. generate the Eclipse project
mvn eclipse:eclipse
9. create the spout of HelloWorldSpout (HelloWorldSpout.java)
package storm.cookbook
class HelloWorldSpout : BaseRichSpout {
private SpoutOutputCollector collector;
private int referenceRandom;
private static final int MAX_RANDOM = 10;
public HelloWorldSpout() {
final Random rand = new Random();
referenceRandom = rand.nextInt(MAX_RANDOM);
}
}
10. after construction, the Storm cluster will open the spout
public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
this.collector = collector;
}
11. The Storm cluster will repeatedly call the nextTuple method, which will do all the work of the spout
public void nextTuple() {
final Random rand = new Random();
int instanceRandom = rand.nextInt(MAX_RANDOM);
if (instanceRandom == referenceRandom) {
collector.emit(new Values("Hello World"));
}
else {
collector.emit(new Values("Other Random Word"));
}
}
12. tell the Storm cluster which fields this spout emits within the declareOutputFields
public void declareOutputFields() {
declarer.declare(new Fields("sentence"));
}
13. create the bolt of HelloWorldBolt (HelloWorldBolt.java)
package storm.cookbook
class HelloWorldBolt : BaseRichBolt {
private int myCount;
public void execute() {
String test = input.getStringByField("sentence");
if ("Hello World".equals(test)) {
myCount++;
System.out.println("Found a Hello World! My Count is now: " + Integer.toString(myCount));
}
}
}
14. create a main class to declare the Storm topology (HelloWorldTopology.java)
package storm.cookbook
public class HelloWorldTopology {
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("randomHelloWorld", new HelloWorldSpout(), 10);
builder.setBolt("HelloWorldBolt", new HelloWorldBolt(), 2).shuffleGrouping("randomHelloWorld");
Config conf = new Config();
conf.setDebug(true);
if(args!=null && args.length > 0) {
conf.setNumWorkers(3);
StormSubmitter.submitTopology(args[0], conf, builder.createTopology());
}
else {
LocalCluster cluster = new LocalCluster();
cluster.submitTopology("test", conf, builder.createTopology());
Utils.sleep(10000);
cluster.killTopology("test");
cluster.shutdown();
}
}
15. execute the cluster from the project's root folder
mvn compile exec:java -Dexec.classpathScope=compile -Dexec.mainClass=storm.cookbook.HelloWorldTopology
Creating a Storm Cluster - provisioning the machines
. create a new project named vagrant-storm-cluster with the following data structure
vagrant-storm-cluster
vagrant-storm-cluster/data
vagrant-storm-cluster/manifest
vagrant-storm-cluster/modules
vagrant-storm-cluster/scripts
. create a file in the project root called Vagrantfile
# -*- mode: ruby -*-
# vi: set ft=ruby :
boxes = [
{ :name => :nimbus, :ip => , :memory => },
{ :name => :supervisor1, :ip => , :memory => },
{ :name => :supervisor2, :ip => , :memory => },
{ :name => :zookeeper1, :ip => , :memory => }
]
. define the hardware, networking, and operating system
boxes.each do |opts|
config.vm.define opts[:name] do |config|
config.vm.box = "ubuntu12"
config.vm.box_url = "http://dl.dropbox.com/u/1537815/precise64.box"
config.vm.network :hostonly, opts[:ip]
config.vm.host_name = "storm.%s" % opts[:name].to_s
config.vm.share_folder "v-data", "/vagrant_data", "./data", :transient => false
config.vm.customize ["modifyvm", :id, "--memory", opts[:memory]]
config.vm.customize ["modifyvm", :id, "--cpus", opts[:cpus]] if opts[:cpus]
. config the provisioning of the application
config.vm.provision :shell, :inline => "cp -fv /vagrant_data/hosts/etc/hosts"
config.vm.provision :shell, :inline => "apt-get update"
# Check if the jdk has been provided
if File.exist?("./data/jdk-6u35-linux-x64.bin") then
config.vm.provision :puppet do |puppet|
puppet.manifests_path = "manifests"
puppet.manifest_file = "jdk.pp"
end
end
config.vm.provision :puppet do |puppet|
puppet.manifests_path = "manifests"
puppet.manifest_file = "provisioningInit.pp"
end
# Ask puppet to do the provisioning now.
config.vm.provision :shell, :inline => "puppet apply /tmp/storm-puppet/manifests/site.pp --verbose --modulepath=/tmp/storm-puppet/modules/ --debug"
end
end
. create installJdk.sh file in the scripts folder
#!/bin/sh
echo "Installing JDK!"
/vagrant_data/jdk-6u35-linux-x64.bin
cd /root
yes | /vagrant_data/jdk-6u35-linux-x64.bin
.0_35 /opt
rm -rv /usr/bin/java
rm -rv /usr/bin/javac
.0_35/bin/java /usr/bin
.0_35/bin/javac /usr/bin
export JAVA_HOME=/opt/jdk1..0_35
export PATH=$PATH:$JAVA_HOME/bin
. create jdk.pp file in the manifest folder
$JDK_VERSION = "1.6.0_35"
package { "openjdk":
ensure => absent,
}
exec { "installJdk":
command => "installJdk.sh",
path => "/bagrant/scripts",
logoutput => true,
creates => "/opt/jdk${JDK_VERSION}",
}
. create provisioningInit.pp file in the manifest folder
$CLONE_URL = "https://bitbucket.org/qanderson/storm-puppet.git"
$CHECKOUT_DIR="/tmp/storm-puppet"
package {git:ensure=> [latest,installed]}
package {puppet:ensure=> [latest,installed]}
package {ruby:ensure=> [latest,installed]}
package {rubygems:ensure=> [latest,installed]}
package {unzip:ensure=> [latest,installed]}
exec { "install_hiera":
command => "gem install hiera hiera-puppet",
path => "/usr/bin",
require => Package['rubygems'],
}
. clone the repository, which contains the second level of provision
exec { "clone_storm-puppet":
command => "git clone ${CLONE_URL}",
cwd => "/tmp",
path => "/usr/bin",
creates => "${CHECKOUT_DIR}",
require => Package['git'],
}
. configure Puppet plugin of Hiera, which is used to externalize properties from the provisioning scripts
exec {"/bin/ln -s /var/lib/gems/1.8/gems/hiera-puppet-1.0.0/ /tmp/storm-puppet/modules/hiera-puppet":
creates => "/tmp/storm-puppet/modules/hiera-puppet",
require => [Exec['clone_storm-puppet'],Exec['install_hiera']]
}
#install hiera and the storm configuration
file { "/etc/puppet/hiera.yaml":
source => "/vagrant_data/hiera.yaml",
replace => true,
require => Package['puppet']
}
file { "/etc/puppet/hieradata":
ensure => directory,
require => Package['puppet']
}
file {"/etc/puppet/hieradata/storm.yaml":
source => "${CHECKOUT_DIR}/modules/storm.yaml",
replace => true,
require => [Exec['clone_storm-puppet'],File['/etc/puppet/hieradata']]
}
. create the Hiera base configuration file in data folder
hiera.yaml:
---
:hierarchy:
- %{operatingsystem}
- storm
:backends:
- yaml
:yaml:
:datadir: '/etc/puppet/hieradata'
. config the host's file
127.0.0.1 localhost
192.168.33.100 storm.nimbus
192.168.33.101 storm.supervisor1
192.168.33.102 storm.supervisor2
192.168.33.103 storm.supervisor3
192.168.33.104 storm.supervisor4
192.168.33.105 storm.supervisor5
192.168.33.201 storm.zookeeper1
192.168.33.202 storm.zookeeper2
192.168.33.203 storm.zookeeper3
192.168.33.204 storm.zookeeper4
. init the Git repository for this project and push it to bitbucket.org
Creating a Storm cluster - provisioning Storm
Once you have a base set of virtual machines that are ready for application provisioning, you need to install and configure the appropriate packages on each node.
. create a new project named storm-puppet
storm-puppet
storm-puppet/manifests
storm-puppet/modules
storm-puppet/modules/storm
storm-puppet/modules/storm/manifests
storm-puppet/modules/storm/templates
. create site.pp in the manifests folder
node 'storm.nimbus' {
$cluster = 'storm1'
include storm::nimbus
include storm::ui
}
node /storm.supervisor[-]/ {
$cluster = 'storm1'
include storm::supervisor
}
node /storm.zookeeper[-]/ {
include storm::zoo
}
. create init.pp in /modules/storm/manifests
class storm {
include storm::install
include storm::config
}
. create install.pp (in /modules/storm/manifests? or /manifests?)
class storm::install {
$BASE_URL="https://bitbucket.org/qanderson/storm-deb-packaging/downloads/"
$ZMQ_FILE="libzmq0_2.1.7_amd64.deb"
$JZMQ_FILE="libjzmq_2.1.7_amd64.deb"
$STORM_FILE="storm_0.8.1_all.deb"
package { "wget":
ensure => latest
}
# call fetch for each file
exec { "wget_storm":
command => "/usr/bin/wget ${BASE_URL}${STORM_FILE}"
}
exec {"wget_zmq":
command => "/usr/bin/wget ${BASE_URL}${ZMQ_FILE}"
}
exec { "wget_jzmq":
command => "/usr/bin/wget ${BASE_URL}${JZMQ_FILE}"
}
#call package for each file
package { "libzmq0":
provider => dpkg,
ensure => installed,
source => "${ZMQ_FILE}",
require => Exec['wget_zmq']
}
#call package for each file
package { "libjzmq":
provider => dpkg,
ensure => installed,
source => "${JZMQ_FILE}",
require => [Exec['wget_jzmq'],Package['libzmq0']]
}
#call package for each file
package { "storm":
provider => dpkg,
ensure => installed,
source => "${STORM_FILE}",
require => [Exec['wget_storm'], Package['libjzmq']]
}
}
. create config.pp in the storm manifests
class storm::config {
require storm::install
include storm::params
file { '/etc/storm/storm.yaml':
require => Package['storm'],
content => template('storm/storm.yaml.erb'),
owner => 'root',
group => 'root',
mode => '
}
file { '/etc/default/storm':
require => Package['storm'],
content => template('storm/default.erb'),
owner => 'root',
group => 'root',
mode => '
}
}
. create params.pp in the storm manifests for Hiera
class storm::params {
#_ STORM DEFAULTS _#
$java_library_path = hiera_array('java_library_path', ['/usr/local/lib', '/opt/local/lib', '/usr/lib'])
}
. specify the nimbus, supervisor, ui, and zoo class
class storm::nimbus {
require storm::install
include storm::config
include storm::params
# Install nimbus /etc/default
storm::service { 'nimbus':
start => 'yes',
jvm_memory => $storm::params::nimbus_mem
}
}
class storm::supervisor {
require storm::install
include storm::config
include storm::params
# Install supervisor /etc/default
storm::service { 'supervisor':
start => 'yes',
jvm_memory => $storm::params::supervisor_mem
}
}
class storm::ui {
require storm::install
include storm::config
include storm::params
# Install ui /etc/default
storm::service { 'ui':
start => 'yes',
jvm_memory => $storm::params::ui_mem
}
}
class storm::zoo {
package {['zookeeper','zookeeper-bin','zookeeperd']:
ensure => latest,
}
}
. init the Git repository and push it to bitbucket.org
. navigate to the vagrant-storm-cluster folder to run the provisioning
vagrant up
. vagrant ssh nimbus
Deriving basic click statistics
Getting ready
wget http://download.redis.io/redis-stable.tar.gz
tar xvzf redis-stable.tar.gz
cd redis-stable
make
sudo cp redis-server /usr/local/bin/
sudo cp redis-cli /usr/local/bin/
Then start the Redis Server.
1. crate a new java project named click-topology, and create the pom.xml file and folder structure as per the "Hello World" topology project.
mkdir ClickTopology
src/test
src/main/java
vi pom.xml
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>storm.cookbook</groupId>
<artifactId>click-topology</artifactId>
<version>0.0.1-SNAPSHOT</version>
<packaging>jar</packaging>
<name>click-topology</name>
<url>https://bitbucket.org/[user]/hello-world</url>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
</project>
2. add the <dependencies> tag
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.11</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.jmock</groupId>
<artifactId>jmock-junit4</artifactId>
<version>2.5.1</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.jmock</groupId>
<artifactId>jmock-legacy</artifactId>
<version>2.5.1</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>redis.clients</groupId>
<artifactId>jedis</artifactId>
<version>2.1.0</version>
</dependency>
3. create the ClickTopology main class in the package of storm.cookbook under the folder of src/main/java
public class ClickTopology {
TopologyBuilder builder = new TopologyBuilder();
public ClickTopology() {
builder.setSpout("clickSpout", new ClickSpout(), 10);
//First layer of bolts
builder.setBolt("repeatsBolt", new RepeatVisitBolt(), 10).shuffleGrouping("clickSpout");
builder.setBolt("geographyBolt", new GeographyBolt(new HttpIPResolver()), 10).shuffleGrouping("clickSpout");
//second layer of bolts, commutative in nature
builder.setBolt("totalStats", new VisitStatsBolt(), 1).globalGrouping("repeatsBolt");
builder.setBolt("geoStats", new GeoStatsBolt(), 10).fieldsGrouping("geographyBolt", new Fields(storm.cookbook.Fields.COUNTRY));
conf.put(Conf.REDIS_PORT_KEY, DEFAULT_JEDIS_PORT);
}
public void runLocal(int runTime){
conf.setDebug(true);
conf.put(Conf.REDIS_HOST_KEY, "localhost");
cluster = new LocalCluster();
cluster.submitTopology("test", conf, builder.createTopology());
if(runTime > 0){
Utils.sleep(runTime);
shutDownLocal();
}
}
public void shutDownLocal(){
if(cluster != null){
cluster.killTopology("test");
cluster.shutdown();
}
}
public void runCluster(String name, String redisHost) throws AlreadyAliveException, InvalidTopologyException {
conf.setNumWorkers(20);
conf.put(Conf.REDIS_HOST_KEY, redisHost);
StormSubmitter.submitTopology(name, conf, builder.createTopology());
}
}
4. add the main method.
public static void main(String[] args) throws Exception {
ClickTopology topology = new ClickTopology();
if(args!=null && args.length > 1) {
topology.runCluster(args[0], args[1]);
} else {
if(args!=null && args.length == 1) {
System.out.println("Running in local mode, redis ip missing for cluster run");
}
topology.runLocal(10000);
}
}
5. the topology assumes that the web server pushes message onto a Redis queue. you must create a spout to inject these into the Storm cluster as a stream. create the ClickSpout class, connect to Redis when it is opened by the cluster.
public class ClickSpout extends BaseRichSpout {
public static Logger logger = Logger.getLogger(ClickSpout.class);
private Jedis jedis;
private String host;
private int port;
private SpoutOutputCollector collector;
@Override
public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
outputFieldsDeclarer.declare(new Fields(storm.cookbook.Fields.IP, storm.cookbook.Fields.URL, storm.cookbook.Fields.CLIENT_KEY));
}
@Override
public void open(Map conf, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
host = conf.get(Conf.REDIS_HOST_KEY).toString();
port = Integer.valueOf(conf.get(Conf.REDIS_PORT_KEY).toString());
this.collector = spoutOutputCollector;
connectToRedis();
}
private void connectToRedis() {
jedis = new Jedis(host, port);
}
}
6. the cluster will then poll the spout for new tuples through the nextTuple method
public void nextTuple() {
String content = jedis.rpop("count");
if (content == null || "nil".equals(content)) {
try {
Thread.sleep(300);
}
catch (InterruptdException e) {
}
}
else {
JSONObject obj = (JSONObject)JSONValue.parse(content);
String ip = obj.get(storm.cookbook.Fields.IP).toString();
String url = obj.get(storm.cookbook.Fields.URL).toString();
String clientKey = obj.get(storm.cookbook.Fields.CLIENT_KEY).toString();
collector.emit(new Values(ip, url, clientKey));
}
}
7. create the bolts that will enrich the basic data through the database or remote API lookups.
public class RepeatVisitBolt extends BaseRichBolt {
private OutputCollector collector;
private Jedis jedis;
private String host;
private int port;
@Override
public void prepare(Map conf, TopologyContext topologyContext, outputCollector outputCollector) {
this.collector = outputCollector;
host = conf.get(Conf.REDIS_HOST_KEY).toString();
port = Integer.valueOf(conf.get(Conf.REDIS_PORT_KEY).toString());
connectToRedis();
}
private void connectToRedis() {
jedis = new Jedis(host, port);
jedis.connect();
}
}
8. add the execute method, which look up the previous visit flags from Redis, based on the fields in the tuple, and emit the enriched tuple
public void execute(Tuple tuple) {
String ip = tuple.getStringByField(storm.cookbook.Fields.IP);
String clientKey = tuple.getStringByField(storm.cookbook.Fields.CLIENT_KEY);
String url = tuple.getStringByField(storm.cookbook.Fields.URL);
String key = url + ":" + clientKey;
String value = jedis.get(key);
if(value == null) {
jedis.set(key, "visited");
collector.emit(new Values(clientKey, url, Boolean.TRUE.toString()));
}
else {
collector.emit(new Values(clientKey, url, Boolean.FALSE.toString()));
}
}
9. create the GeographyBolt class
package storm.bookbook;
public class GeographyBolt extends BaseRichBolt {
public void execute(Tuple tuple) {
String ip = tuple.getStringByField(storm.cookbook.Fields.IP);
JSONObject json = resolver.resolveIP(ip);
String city = (String) json.get(storm.cookbook.Fields.CITY);
String country = (String) json.get(storm.cookbook.Fields.COUNTRY_NAME);
collector.emit(new Values(country, city));
}
}
10. create the HttpIPResolver class, and injecting it into GeographyBolt at design time
public class HttpIPResolver implements IPResolver, Serializable {
}
11. splits streams into the bolt
builder.setBolt("geoStats", new GeoStatsBolt(), 10).fieldsGrouping("geographyBolt", new Fields(storm.bookbook.Fields.COUNTRY));
12. create the GeoStatsBolt class
public class GeoStatsBolt {
public void execute(Tuple tuple) {
String country = tuple.getStringByField(storm.cookbook.Fields.COUNTRY);
String city = tuple.getStringByField(Fields.CITY);
if(!stats.containsKey(country)){
stats.put(country, new CountryStats(country));
}
stats.get(country).cityFound(city);
collector.emit(new Values(country, stats.get(country).getCountryTotal(), city, stats.get(country).getCityTotal(city)));
}
}
13. create the CountryStats class
private class CountryStatus {
private int countryTotal = 0;
private static final int COUNT_INDEX = 0;
private static final int PERCENTAGE_INDEX = 1;
private String countryName;
private Map<String, List<Integer>> cityStats = new HashMap<String, List<Integer>>();
public CountryStats(String countryName) {
this.countryName = countryName;
}
public void cityFound(String cityName) {
countryTotal++;
if(cityStats.containsKey(cityName)){
cityStats.get(cityName).set(COUNT_INDEX, cityStats.get(cityName).get(COUNT_INDEX).intValue() + 1);
}
else {
List<Integer> list = new LinkedList<Integer>();
list.add(1);
list.add(0);
cityStats.put(cityName, list);
}
double percent = (double)cityStats.get(cityName).get(COUNT_INDEX)/(double)countryTotal;
cityStats.get(cityName).set(PERCENTAGE_INDEX, (int)percent);
}
public int getCountryTotal(){
return countryTotal;
}
public int getCityTotal(String cityName){
return cityStats.get(cityName).get(COUNT_INDEX).intValue();
}
}
14. the final counting for visitors and unique visitors
builder.setBolt("totalStats", new VisitStatsBolt(), 1).globalGrouping("repeatsBolt");
15. create the VisitStatsBolt class
public class VisitStatsBolt {
public void execute(Tuple tuple) {
boolean unique = Boolean.parseBoolean(tuple.getStringByField(storm.cookbook.Fields.UNIQUE));
total++;
if(unique) {
uniqueCount++;
}
collector.emit(new Values(total,uniqueCount));
}
}
Unit testing a bolt
1. create the StormTestCase class under src/text/java
package storm.cookbook;
public class StormTestCase {
protected Mockery context = new Mockery() {
setImposteriser(ClassImposteriser.INSTANCE);
};
protected Tuple getTuple() {
final Tuple tuple = context.mock(Tuple.class);
return tuple;
}
}
2. craete the TestRepeatVisitBolt class
@RunWith(value = Parameterized.class)
public class TestRepeatVisitBolt extends StormTestCase {
}
3. add the execute method
public void testExecute() {
jedis = new Jedis("localhost", 6379);
RepeatVisitBolt bolt = new RepeatVisibleBolt();
Map config = new HashMap();
config.put("redis-host", "localhost");
config.put("redis-port", "6379");
final OutputCollector collector = context.mock(OutputCollector.class);
bolt.prepare(config, null, collector);
final Tuple tuple = getTuple();
context.checking(new Expectations() {
oneOf(tuple).getStringByField(Fields.IP);
will(returnValue(ip));
oneOf(tuple).getStringByField(Fields.CLIENT_KEY);
will(returnValue(clientKey));
oneOf(tuple).getStringByField(Fields.URL);
will(returnValue(url));
oneOf(collector).emit(new Values(clientKey, url, expected));
});
bolt.execute(tuple);
context.assertIsSatisfied();
if(jedis != null) {
jedis.disconnect();
}
}
4. define the parameters
@Parameterized.Parameters
public static Collection<Object[]> data() {
Object[][] data = new Object[][] {
{ "192.168.33.100", "Client1", "myintranet.com", "false" },
{ "192.168.33.100", "Client1", "myintranet.com", "false" },
{ "192.168.33.101", "Client2", "myintranet1.com", "true" },
{ "192.168.33.102", "Client3", "myintranet2.com", false"}
};
return Arrays.asList(data);
}
5. add the base provisioning of the values using Redis
@BeforeClass
pubic static void setupJedis() {
Jedis jedis = new Jedis("localhost",6379);
jedis.flushDB();
Iterator<Object[]> it = data().iterator();
while (it.hasNext()) {
Object[] values = it.next();
if (values[3].equals("false")) {
String key = values[2] + ":" + values[1];
jedis.set(key, "visited");
}
}
}
create storm topologies and deploy to storm cluster.
Storm cluster <similar to> Hadoop cluster
topologies <whereas> MapReduce jobs
process forever < whereas> eventually finishes
master node => worker node
daemon<Nimbus> => JobTracker
distributing code
daemon<Supervisor> =>
storm jar all-my-code.jar org.apache.storm.MyTopology arg1 arg2
storm jar part takes care of connecting to Nimbus and uploading the jar.
Streams
=> unbounded sequence of tuples.
stream transformation => spouts and bolts, have interfaces
spout => source of stream
bolt => consume input stream, process and emit new stream
=> run functions, filter tuples, do streaming aggregation, do streaming join, talk to database
tuple => data model