Showing posts with label BigData. Show all posts
Showing posts with label BigData. Show all posts

Thursday, March 31, 2016

Setup Hadoop YARN on CentOS VMs

This post summarizes my experience in setting up testing environment for YARN using CentOS  VMs.

After we setup the hdfs following the link (http://czcodezone.blogspot.sg/2016/03/setup-hdfs-cluster-in-centos-vms.html). We can go ahead to set up yarn to manage jobs in hadoop.

Hadoop v2 uses application masters in the datanode to work together with nodemanager to manage a job, and uses resource manager in the namenode to schedule resources for a job (a job refers to a particular application or driver from distributed computation framework such as mapreduce or spark).

1. Setup yarn configuration in hadoop


To setup yarn, on each VM (both namenode and datanode), perform the following steps:

1.1. Edit hadoop/etc/hadoop/mapred-site.xml


Run the following command to edit the mapred-site.xml:

```bash
`cd hadoop/etc/hadoop
`cp mapred-site.xml.template mapred-site.xml
`vi mapred-site.xml

In the mapred-site.xml, Modify as follows:

<configuration>
<property>
<name>fs.default.name</name>
<value>hdfs://centos01:9000</value>
</property>
</configuration>

The "hdfs://centos01:9000" specify the master as the "fs.default.name". (It is important that this is not specified as "hdfs://localhost:9000", otherwise the "hadoop/bin/hdfs dfsadmin -report" will have "Connection refused" exception)


1.2 Edit hadoop/etc/hadoop/yarn-site.xml


Run the following command to edit the yarn-site.xml:

```bash
`vi hadoop/etc/hadoop/yarn-site.xml

In the yarn-site.xml, modify as follows:

<configuration>
<property>
<name>yarn.nodemanager.aux-services</name>
<value>mapreduce_shuffle</value>
</property>
<property>
<name>yarn.nodemanager.aux-services.mapreduce_shuffle.class</name>
<value>org.apache.hadoop.mapred.ShuffleHandler</value>
</property>
</configuration>

2. Start the yarn in the namenode.


On the namenode centos01 (please refers to http://czcodezone.blogspot.sg/2016/03/setup-centos-vm-in-virtualbox-for.html), run the following command to start the hdfs and then yarn:

```bash
`hadoop/sbin/start-dfs.sh
`hadoop/sbin/start-yarn.sh
`hadoop/sbin/mr-jobhistory-daemon.sh start historyserver
`jps

3. Stop hdfs and yarn


To stop hdfs and yarn, run the following command:

```bash
`hadoop/sbin/mr-jobhistory-daemon.sh stop historyserver
`hadoop/sbin/stop-yarn.sh
`hadoop/sbin/stop-dfs.sh

Setup Spark Cluster in CentOS VMs

This post summarizes my experience in setting up a test environment for spark cluster using CentOS VMs.

After setup the VMs, we will designate the following topology (refer to this for setting up and configure CentOS VMs using VirtualBox http://czcodezone.blogspot.sg/2016/03/setup-centos-vm-in-virtualbox-for.html):

centos01: master
centos02: slave
centos03: slave
centos04: slave
centos05: slave

where centos01 refers to the hostname of the VM.

1. Configure the spark cluster


Download and unzip the spark-1.6.0-hadoop2.6.tgz to "/root/spark", run the following command on centos01 to specify the list of slaves:

```bash
`cp spark/conf/slaves.template spark/conf/slaves
`vi spark/conf/slaves

In the spark/conf/slaves add the following lines:

centos02
centos03
centos04
centos05

Make sure that the firewalls turned off on centos01/2/3/4/5 and and passwordless ssh from centos01 to centos02/3/4/5.

Run the following command to copy the "/root/spark" from centos01 to centos02/3/4/5:

```bash
`rsync -a /root/spark/ root@centos02:/root/park
`rsync -a /root/spark/ root@centos03:/root/park
`rsync -a /root/spark/ root@centos04:/root/park
`rsync -a /root/spark/ root@centos05:/root/park

2. Start and stop the spark cluster


Run the following command on centos01 to start the spark cluster:

```bash
`spark/sbin/start-all.sh


To stop the spark cluster, run the following comand on centos01:

```bash
`spark/sbin/stop-all.sh

3. Run the spark shell


After the spark cluster is started, we can start the spark shell by running the following command:

```bash
`spark/bin/spark-shell --master spark://centos01:7077

The port 7077 is the default port for spark master centos01

4. Submit a spark job to spark cluster


After the spark cluster has been setup, assuming the master is centos01, run the following command to submit a spark job:

```bash
`spark/bin/spark-submit --class com.tutorials.spark.WordCountDriver --master spark://centos01:7077 word-count.jar

Or more refinely:
```bash
`spark/bin/spark-submit --class com.tutorials.spark.WordCountDriver --master spark://centos01:7077 --executor-memory 2G --total-executor-cores 8 word-count.jar 1000

Below are two other alternatives i tested using YARN cluster and mesos cluster

4.1. Submit a spark job via YARN cluster

Suppose we have a resource management cluster such as Hadoop YARN setup, we can submit the spark job to YARN for processing as well (YARN will the spark master)

To run an application in YARN cluster instead, setup and configure hdfs and yarn using the (link: http://czcodezone.blogspot.sg/2016/03/setup-hdfs-cluster-in-centos-vms.html) and (link: http://czcodezone.blogspot.sg/2016/03/setup-hadoop-yarn-on-centos-vms.html), Start the HDFS and YARN.

run the following command to edit the .bashrc:

```bash
`vi .bashrc

In the spark/conf/spark-env.sh of each VM centos01/2/3/4/5, add the following line:

export HADOOP_HOME_DIR=/root/hadoop
export HADOOP_CONF_DIR=/root/hadoop/etc/hadoop
export HADOOP_YARN_DIR=/root/hadoop/etc/hadoop

Run the following command on each VM to update .bashrc:

```bash
`source .bashrc

To submit a spark job, run the following command:

```bash
``spark/bin/spark-submit  --class com.tutorials.spark.WordCountDriver --master yarn --deploy-mode cluster word-count.jar

Or
```bash
``spark/bin/spark-submit  --class com.tutorials.spark.WordCountDriver --master yarn-cluster word-count.jar

Or more refinely:
```bash
`spark/bin/spark-submit  --class com.tutorials.spark.WordCountDriver --master yarn --deploy-mode cluster --executor-memory 2G --num-executors 20 word-count.jar 1000

4.2. Submit a spark job via mesos cluster


To run an application in mesos cluster instead, setup and configure hdfs and mesos (link: http://czcodezone.blogspot.sg/2016/03/setup-mesos-cluster-in-centos-vms.html), Start the HDFS and MESOS.

Put the spark bin package in the hdfs (run the following command on the hadoop namenode centos01):

```bash
`wget http://www.apache.org/dyn/closer.lua/spark/spark-1.6.0/spark-1.6.0-bin-hadoop2.6.tgz
`hadoop/bin/hdfs dfs -mkdir /pkg
`hadoop/bin/hdfs dfs -put spark-1.6.0-bin-hadoop2.6.tgz /pkg/spark-1.6.0-bin-hadoop2.6.tgz

Run the following command to modify the spark-env.sh in spark/conf

```bash
`vi spark/conf/spark-env.sh

In the spark-env.sh, add the following lines:

export MESOS_NATIVE_LIBRARY=/usr/local/lib/libmesos.so
export SPARK_EXECUTOR_URI= hdfs://centos01:9000/pkg/spark-1.6.0-bin-hadoop2.6.tgz

Where centos01 is the hadoop namenode

To submit a spark job, run the following command:

```bash
`spark/bin/spark-submit --class com.tutorials.spark.WordCountDriver --master mesos://mesos01:5050 word-count.jar

Important: mesos01 must be the current leader master node, otherwise, the command such as "spark-shell --master mesos://mesos01:5050" will cause the spark-shell to hang on the line "No credential provided. attempting to register without authentication". The solution is to find out which node is the active leader master node by running the command "mesos-resolve `cat /etc/mesos/zk`" and the luanch the spark shell by specifying the active leader master as in the --master option instead.

Setup HDFS Cluster in CentOS VMs

This post summarizes my experience in setting up a test environment for HDFS cluster using CentOS VMs.

Before we start we like to configure the VMs (Refer to this on how to setup and configure CentOS VMs using VirtualBox for HDFS: http://czcodezone.blogspot.sg/2016/03/setup-centos-vm-in-virtualbox-for.html) to be the following:

centos01/192.168.56.101: run namenode
centos02/192.168.56.102: run datanode
centos03/192.168.56.103: run datanode
centos04/192.168.56.104: run datanode
centos05/192.168.56.105: run datanode

where centos01 is the hostname and the 192.168.56.101 is the ip assigned to host centos01

1. Set up hostname for each computer

In this case, we assume that we have not set up a DNS  for the VMs to know each other, but we do not want to use raw ip addresses. Therefore we need to configure VMs to identify by hostname centos0x.

On  each VM above, do the following (in the following case, we use centos01/192.168.56.101 for illustration, DO replace them with the individual VM's hostname and ipaddress instead).

1.1. Modify /etc/sysconfig/network


Run the following command to edit /etc/hostname:

```bash
`vi /etc/hostname

In the /etc/hostname, put the following line (replace "centos01" accordingly)

centos01

Run the following command to edit /etc/sysconfig/network:

```bash
`vi /etc/sysconfig/network

In the /etc/sysconfig/network, put the following line (replace "centos01" accordingly)

HOSTNAME=centos01

Run the following command to restart network service:

```bash
`service network restart

1.2. Modify /etc/hosts


Run the following command to edit /etc/hosts:

```bash
`vi /etc/hosts

In the /etc/hosts, add the following lines

192.168.56.101 centos01
192.168.56.102 centos02
192.168.56.103 centos03
192.168.56.104 centos04
192.168.56.105 centos05

2. Set up passwordless ssh from the namenode to the datanodes


We need to setup passwordless ssh from the namenode (namely centos01) to the rest of the VMs which serve as datanodes and itself. To do this, on the centos01, run the following command to create the id_dsa.pub public key:

```bash
`mkdir ~/.ssh
`ssh-keygen -t dsa -P '' -f ~/.ssh/id_dsa
`scp ~/.ssh/id_dsa.pub root@centos02:/root
`scp ~/.ssh/id_dsa.pub root@centos03:/root
`scp ~/.ssh/id_dsa.pub root@centos04:/root
`scp ~/.ssh/id_dsa.pub root@centos05:/root
`cat ~/.ssh/id_dsa.pub >> ~/.ssh/authorized_keys

Now the id_dsa.pub has been copied from centos01 to the other 4 VMs's root directory, we need to append them to /root/.ssh/authorized_keys. for each VM in centos02/3/4/5, run the following command:

```bash
`mkdir ~/.ssh
`touch ~/.ssh/authorized_keys
`cat ~/id_dsa.pub >> ~/.ssh/authorized_keys

3. Configure hadoop on each VMs


Perform the following steps on each VM of centos01/2/3/4/5

3.1 Configure $JAVA_HOME 

Before we start running hdfs, we need to specify the JAVA_HOME in the environment path. Assume that we install the java-1.8.0-openjdk-devel on each VM for java jdk, the installation is at /usr/lib/jvm/java-1.8.0-openjdk. To specify the JAVA_HOME in the environment path. run the following command:

```bash
`vi ~/.bashrc

In the .bashrc, add the following line:

export JAVA_HOME=/usr/lib/jvm/java-1.8.0-openjdk

3.2. Download hadoop 


Download the hadoop binary distribution, unzip it to the root directory ~/hadoop.

3.3 Configure ~/hadoop/etc/hadoop/slaves 


Run the folowing command to edit slaves:

```bash
`vi ~/hadoop/etc/hadoop/slaves

In the ~/hadoop/etc/hadoop/slaves, remove the "localhost" and add the following line:

centos02
centos03
centos04
centos05

This slaves files specified the datanodes

3.4. Configure ~/hadoop/etc/hadoop/core-site.xml


Run the following command to edit core-site.xml:

```bash
`vi ~/hadoop/etc/hadoop/core-site.xml

In ~/hadoop/etc/hadoop/core-site.xml, write the following (centos01 refers to the namenode):

<configuration>
<property>
<name>fs.defaultFS</name>
<value>hdfs://centos01:9000/</value>
</property>
</configuration>

3.5. Configure ~/hadoop/etc/hdfs-site.xml


Run the following command to create directory for hdfs data:

```bash
`mkdir ~/hadoop_data
`mkdir ~/hadoop_data/data
`mkdir ~/hadoop_data/name
`mkdir ~/hadoop_data/local
`chmod -R 755 ~/hadoop_data

In ~/hadoop/etc/hadoop/hdfs-site.xml, write the following:

<configuration>
<property>
<name>dfs.datanode.data.dir</name>
<value>/root/hadoop_data/data</value>
</property>
<property>
<name>dfs.namenode.name.dir</name>
<value>/root/hadoop_data/name</value>
</property>
</configuration>

The above specify how the data for namenode and datanode are stored.

4. Start the hdfs cluster


On the namenode centos01, run the following command to format namenode:

```bash
`~/hadoop/bin/hdfs namenode -format

On the namenode centos01, start the hdfs cluster:

```bash
`~/hadoop/sbin/start-dfs.sh

To check what are running on each VM, run the following command on each VM:

```bash
`jps

To check the reporting of the hdfs cluster, run the following command on the namenode centos01:

```bash
`~/hadoop/bin/hdfs dfsadmin -report

Another way to check is to visit the web server hosted by namenode:

```bash
`curl http://centos01:50070/

5. Stop the hdfs cluster


To stop the hdfs cluster, run the following commmand on the namenode centos01:

```bash
`~/hadoop/sbin/stop-dfs.sh


Monday, February 1, 2016

Elasticsearch Version Upgrade using Rolling Restart

This upgrade method follows updating one node at a time and restart again. The same method can also be used for restarting a elasticsearch cluster in a safe and efficient way.

Master-eligible nodes


In a elasticsearch production cluster, start with one master-eligible node, stop the elasticsearch service, perform upgrade, and then restart the service. Repeat this until all master-eligible nodes have been upgraded. Since master-eligible nodes do not keep shards and replicas, the process is safe with data so far.

Client nodes


Next stop, upgrade, and restart the client nodes one at a time just like with the master-eligible nodes.

Data nodes


Next, before starting to upgrade any data node in the elasticsearch cluster, dynamically adding a setting to temporarily turn off the resharding of the elasticsearch cluster via restful api calls to the cluster (because if we restart a data node, the shards in the cluster will rebalance). The setting to be temporally disabled is the cluster.routing.allocation.enable, which can be done by issue the following call to the elasticsearch cluster:

curl -X PUT -H "Content-Type: application/json" http://elastic-cluster:9200/_cluster/settings/ -d '
{ "transient": { "cluster.routing.allocation.enable": "none" } }
'
(Note that the "transient" allows the setting to be not permanent)

Now stop, upgrade and restart a data node. At this point, we can reverse the setting for cluster.routing.allocation.enable by running the curl restful below:

curl -X PUT -H "Content-Type: application/json" http://elastic-cluster:9200/_cluster/settings/ -d '
{ "transient": { "cluster.routing.allocation.enable": "all" } }
'
Once this is done, the origin shards for that data node will be up again.

Next, proceed to the second data node and repeat the process above until all data nodes are upgraded.



Sunday, January 31, 2016

Quorum and minimum_master_nodes setting for elasticsearch configuration

Theory behind elasticsearch recovery

Elastic search recovery works by having a master election satifying a particular minimum_master_nodes criteria. That is a master node will only be elected if there is a N number of master-eligible nodes (nodes in which node.master=true in their elasticsearch.yml file) to join it. N is specified by discovery.zen.minimum_master_nodes in the elasticsearch.yml file.

For  the quorum scheme, N should be set to

discovery.zen.minimum_master_nodes = (number of master-eligible nodes) / 2 + 1

The recovery works like this. When the current master node dies, a new master node will only be elected if there is N master-eligible nodes to join it, where N = (number of master-eligible nodes) / 2 + 1

Once the new master node is elected, if later the originally dead master node comes alive. It will have less than N master-eligible nodes to join it, therefore it will have to step down. Thus this scheme ensures that there will no two masters at the same time (which is the so-called split-brain scenario, that is not desirable since all master nodes have higher authority in its being able to update cluster-state in the data nodes and client nodes)

Minimum number of master eligible nodes required for an elasticsearch cluster

The minimum number of master-eligible nodes should be 3. And the zen.discovery.minimum_master_nodes should be equal = 3 / 2 + 1 = 2.

Reason: Suppose we only have two master-eligible nodes. If the master node dies, there is only 1 master-eligible node left, two things will happen depending on the value in discovery.zen.minimum_master_nodes:

Case 1:  if the zen.discovery.minimum_master_nodes is set to greater than 1, then there won't be any new master node elected, and the cluster will not operate.


Case 2:  if we set the zen.discovery.minimum_master_nodes=1, the new master node will be elected, however, when the originally dead master is brought alive again, the original master will not step down since it now also has one master-eligible node to join it, leading to split-brain problem.

Therefore the recommended minimum settings is to increases the number of master-eligible nodes to 3, and set the zen.discovery.minimum_master_nodes=2.

Wednesday, November 26, 2014

Trident-ML: Indexing Sentiment Classification results with ElasticSearch

In Storm, we may have scenarios in which we like to index results obtained from real-time processing or machine learning into a search and analytics engine. For example, we may have some text streaming in from Kafka messaging system which will go through a TwitterSentimentClassifier (which is available in Trident-ML). After that, we may wish to save the text together with the classified sentiment label as an indexed document in ElasticSearch. This post shows one way to realize such an implementation.

First create a Maven project (e.g. with groupId="com.memeanalytics" and artifactId="es-create-index"), the complete source code of the project can be downloaded from the link below:

https://dl.dropboxusercontent.com/u/113201788/storm/es-create-index.tar.gz

Configure pom.xml and libraries to be used

Before we proceed, I would like to discuss how to write a elasticsearch client which is compatible with Trident-ML as we will be using both in this project. Traditionally an elasticsearch java client can be implemented using native code such as this:

import static org.elasticsearch.node.NodeBuiler.*;

Node node=nodeBuilder().clusterName("elasticsearch").node();
Client client=node.getClient();

//TODO HERE: put document, delete document, etc using the client

node.close();

The above code requires the following dependency in pom.xml:

<dependency>
<groupId>org.elasticsearch</groupId>
<artifactId>elasticsearch</artifactId>
<version>1.4.0</version>
</dependency>

However, as the pom and coding of this library has dependency on lucene-core [version=3.6.0] that is an older version that is not compatible with lucene-analyzers [version=3.6.2] which is currently one of Trident-ML's dependency (The TwitterTokenizer in TwitterSentimentClassifier uses this library). As a result, the elasticsearch library above cannot be used if the TwitterSentimentClassifier in Trident-ML is to be used in this project.

Since the above java code and elastic library cannot be used in this project, the project uses httpclient [version=4.3] from org.apache.httpcomponents in its place to communicate with elasticsearch via RESTful api. The httpclient provides CloseableHttpClient and operators such as HttpGet, HttpPut, HttpDelete,

The dependencies section of the pom for this project looks like the following:

<dependency>
  <groupId>storm</groupId>
  <artifactId>storm</artifactId>
  <version>0.9.0.1</version>
</dependency>
<dependency>
  <groupId>com.github.pmerienne</groupId>
  <artifactId>trident-ml</artifactId>
  <version>0.0.4</version>
  <scope>provided</scope>
</dependency>
<dependency>
  <groupId>org.apache.httpcomponents</groupId>
  <artifactId>httpclient</artifactId>
  <version>4.3</version>
</dependency>

Spout

Once the pom.xml is properly updated, we can move to implement the code for the Storm spout used in this project. The spout, named TweetCommentSpout, reads tweets from "src/test/resources/twitter-sentiment.csv" and emits them in batch to the Trident topology. the implementation of the spout is shown below:

package com.memeanalytics.es_create_index;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import backtype.storm.task.TopologyContext;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;

import com.github.pmerienne.trident.ml.core.TextInstance;
import com.github.pmerienne.trident.ml.preprocessing.EnglishTokenizer;
import com.github.pmerienne.trident.ml.preprocessing.TextTokenizer;

import storm.trident.operation.TridentCollector;
import storm.trident.spout.IBatchSpout;

public class TweetCommentSpout implements IBatchSpout {
 
 private static final long serialVersionUID = 1L;
 private static List<List<Object>> data=new ArrayList<List<Object>>();

 private int batchIndex;
 private int batchSize=10;
 
 static{
  BufferedReader br=null;
  FileInputStream is=null;
  String filePath="src/test/resources/twitter-sentiment.csv";
  try {
   is=new FileInputStream(filePath);
   br=new BufferedReader(new InputStreamReader(is));
   String line=null;
   while((line=br.readLine())!=null)
   {
    String[] values = line.split(",");
    Integer label=Integer.parseInt(values[0]);
    String text=values[1];
//    TextTokenizer tokenizer=new EnglishTokenizer();
//    List<String> tokens = tokenizer.tokenize(text);
//    TextInstance<Integer> instance=new TextInstance<Integer>(label, tokens);
    data.add(new Values(text, label));
   }
  } catch (FileNotFoundException e) {
   e.printStackTrace();
  }catch(IOException ex)
  {
   ex.printStackTrace();
  }
  
 }
 public void open(Map conf, TopologyContext context) {
  // TODO Auto-generated method stub
  
 }

 public void emitBatch(long batchId, TridentCollector collector) {
  // TODO Auto-generated method stub
  int maxBatchCount = data.size() / batchSize;
  if(maxBatchCount > 0 && batchIndex < maxBatchCount)
  {
   for(int i=(batchSize * batchIndex); i < data.size() && i < (batchIndex+1) * batchSize; ++i)
   {
    collector.emit(data.get(i));
   }
   batchIndex++;
  }
 }

 public void ack(long batchId) {
  // TODO Auto-generated method stub
  
 }

 public void close() {
  // TODO Auto-generated method stub
  
 }

 public Map getComponentConfiguration() {
  // TODO Auto-generated method stub
  return null;
 }

 public Fields getOutputFields() {
  // TODO Auto-generated method stub
  return new Fields("text", "label");
 }
 
}

The tuples emitted by the spout contains two fields: "text" and "label", the label is ignored, we are going to have the Trident-ML's TweetSentimentClassifier predict the sentiment label for us instead.

Trident operation for ElasticSearch

Next we are going to implement a BaseFilter, named CreateESIndex, which is a Trident operation that create an indexed document in ElasticSearch from each tweet text and its predicted sentiment label. The implementation of the Trident operation is shown below:

package com.memeanalytics.es_create_index;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;
import java.util.Map;

import org.apache.http.HttpEntity;
import org.apache.http.client.ClientProtocolException;
import org.apache.http.client.HttpRequestRetryHandler;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpDelete;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.client.methods.HttpPut;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.message.BasicHeader;
import org.apache.http.protocol.HTTP;
import org.apache.http.protocol.HttpContext;
import org.apache.http.util.EntityUtils;

import storm.trident.operation.BaseFilter;
import storm.trident.operation.TridentOperationContext;
import storm.trident.tuple.TridentTuple;

public class CreateESIndex extends BaseFilter{

 private static final long serialVersionUID = 1L;
 private int esIndex=1;
 private String wsUrl="http://127.0.0.1:9200";
 private String indexName="twittersentiment"; //must be lowercase
 private String typeName="trident";
 private CloseableHttpClient client;
 private String lastIndexedDocumentIdQueryJson="{\"query\": { \"match_all\": {}}, \"size\": 1,"+
   "\"sort\": ["+
     "{"+
       "\"_timestamp\": {"+
         "\"order\": \"desc\""+
       "}"+
     "}"+
   "]"+
 "}";

 public boolean isKeep(TridentTuple tuple) {
  // TODO Auto-generated method stub
  Boolean prediction =tuple.getBooleanByField("prediction");
  String comment=tuple.getStringByField("text");
  System.out.println(comment + " >> " + prediction);
  
  if(client != null)
  {
   HttpPut method=new HttpPut(wsUrl+"/"+indexName+"/"+typeName+"/"+esIndex);
   
   Date currentTime= new Date();
   SimpleDateFormat format1 = new SimpleDateFormat("yyyy-MM-dd");
   SimpleDateFormat format2 = new SimpleDateFormat("HH:mm:ss");
   String dateString = format1.format(currentTime)+"T"+format2.format(currentTime);
   
   CloseableHttpResponse response=null;
   try{
    String json = "{\"text\":\""+comment+"\", \"prediction\":\""+prediction+"\", \"postTime\":\""+dateString+"\"}";
    System.out.println(json);
    StringEntity params=new StringEntity(json);
    params.setContentType(new BasicHeader(HTTP.CONTENT_TYPE, "application/json"));
    
    method.setEntity(params);
    
    method.addHeader("Accept", "application/json");
    method.addHeader("Content-type", "application/json");
    
    response = client.execute(method);
    
    HttpEntity entity=response.getEntity();
    String responseText=EntityUtils.toString(entity);
    System.out.println(responseText);
   }catch(IOException ex) {
    ex.printStackTrace();
   }finally {
    method.releaseConnection();
   }
   esIndex++;
  }
  
  return true;
 }
 
 @Override
    public void prepare(Map conf, TridentOperationContext context) {
  
  client=HttpClients.custom().setRetryHandler(new MyRetryHandler()).build();
  
  CloseableHttpResponse response=null;
  HttpDelete method=new HttpDelete(wsUrl+"/"+indexName);
  try{
   response = client.execute(method);
   HttpEntity entity=response.getEntity();
   String responseBody=EntityUtils.toString(entity);
   System.out.println(responseBody);
  }catch(IOException ex)
  {
   ex.printStackTrace();
  }
    }
 
 private class MyRetryHandler implements HttpRequestRetryHandler {

  public boolean retryRequest(IOException arg0, int arg1, HttpContext arg2) {
   // TODO Auto-generated method stub
   return false;
  }
 }

    @Override
    public void cleanup() {
     try {
   client.close();
  } catch (IOException e) {
   // TODO Auto-generated catch block
   e.printStackTrace();
  }
    }

}

In the prepare() method of the CreateESIndex, a RESTful DELETE call is performed to delete all indexed documents under twittersentiment/trident in ElasticSearch. This is to ensure that no data will be under twittersentiment/trident when the bolt is run. Now in its isKeep() method, the tweet text and its associated predicted sentiment label is serialized to a json and sent to elasticsearch via a http PUT call. The CloseableHttpClient object is closed in its cleanup() method.

Trident topology

Now we have the neccessary spout and trident operation, we can define a simple Trident topology which stream tweets-> classified by TwitterSentimentClassifier -> indexed by ElasticSearch. Below is the implementation in the main class:

package com.memeanalytics.es_create_index;

import com.github.pmerienne.trident.ml.nlp.TwitterSentimentClassifier;

import storm.trident.TridentTopology;
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.generated.StormTopology;
import backtype.storm.tuple.Fields;

public class App 
{
    public static void main( String[] args )
    {
        LocalCluster cluster=new LocalCluster();
        Config config=new Config();
        
        cluster.submitTopology("TridentWriteToESDemo", config, buildTopology());
        
        try{
         Thread.sleep(10000);
        }catch(InterruptedException ex)
        {
         
        }
        
        cluster.killTopology("TridentWriteToESDemo");
        cluster.shutdown();
    }
    
    private static StormTopology buildTopology()
    {
     TridentTopology topology=new TridentTopology();
     
     TweetCommentSpout spout=new TweetCommentSpout();
     
     topology.newStream("classifyAndIndex", spout).each(new Fields("text"), new TwitterSentimentClassifier(), new Fields("prediction")).each(new Fields("text", "prediction"), new CreateESIndex());
     
     return topology.build();
    }
}

Once it is completed, run the following command in the project root folder:

> mvn compile exec:java

Tuesday, November 25, 2014

Trident-ML: Regression using Passive-Aggressive algorithm

This post shows some very basic example of how to use the Passive-Aggressive algorithm as regression algorithm in Trident-ML to process data from Storm Spout.

Firstly create a Maven project (e.g. with groupId="com.memeanalytics" artifactId="trident-regression-pa"). The complete source codes of the project can be downloaded from the link:

https://dl.dropboxusercontent.com/u/113201788/storm/trident-regression-pa.tar.gz

For the start we need to configure the pom.xml file in the project.

Configure pom.xml:
Firstly we need to add the clojars repository to the repositories section:

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

Next we need to add the storm dependency to the dependencies section (for storm):

<dependency>
  <groupId>storm</groupId>
  <artifactId>storm</artifactId>
  <version>0.9.0.1</version>
  <scope>provided</scope>
</dependency>

Next we need to add the strident-ml dependency to the dependencies section (for PA regression):

<dependency>
  <groupId>com.github.pmerienne</groupId>
  <artifactId>trident-ml</artifactId>
  <version>0.0.4</version>
</dependency>

Next we need to add the exec-maven-plugin to the build/plugins section (for execute the Maven project):

<plugin>
<groupId>org.codehaus.mojo</groupId>
<artifactId>exec-maven-plugin</artifactId>
<version>1.2.1</version>
<executions>
<execution>
<goals>
<goal>exec</goal>
</goals>
</execution>
</executions>
<configuration>
<includeProjectDependencies>true</includeProjectDependencies>
<includePluginDependencies>false</includePluginDependencies>
<executable>java</executable>
<classpathScope>compile</classpathScope>
<mainClass>com.memeanalytics.trident_regression_pa.App</mainClass>
</configuration>
</plugin>

Next we need to add the maven-assembly-plugin to the build/plugins section (for packacging the Maven project to jar for submitting to Storm cluster):

<plugin>
<artifactId>maven-assembly-plugin</artifactId>
<version>2.2.1</version>
<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>

Implement Spout for training data 

Once the pom.xml update is completed, we can move to implement the BirthDataSpout which is the Storm spout that emits batches of training data to the Trident topology:

package com.memeanalytics.trident_regression_pa;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;

import com.github.pmerienne.trident.ml.core.Instance;
import com.github.pmerienne.trident.ml.testing.data.Datasets;

import backtype.storm.task.TopologyContext;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
import storm.trident.operation.TridentCollector;
import storm.trident.spout.IBatchSpout;

public class BirthDataSpout implements IBatchSpout {

 private static final long serialVersionUID = 1L;

 private int batchSize=10;
 private int batchIndex=0;
 
 private static List<Instance<Double>> sample_data=new ArrayList<Instance<Double>>();
 private static List<Instance<Double>> testing_data=new ArrayList<Instance<Double>>();
 
 public static List<String> getDRPCArgsList()
 {
  List<String> drpc_args_list =new ArrayList<String>();
  for(Instance<Double> instance : testing_data)
  {
   double[] features = instance.getFeatures();
   String drpc_args="";
   for(int i=0; i < features.length; ++i)
   {
    if(i==0)
    {
     drpc_args+=features[i];
    }
    else
    {
     drpc_args+=(","+features[i]);
    }
   }
   drpc_args+=(","+instance.label);
   drpc_args_list.add(drpc_args);
  }
  
  return drpc_args_list;
 }
 
 static{
  FileInputStream is=null;
  BufferedReader br=null;
  try{
   String filePath="src/test/resources/births.csv";
   is=new FileInputStream(filePath);
   br=new BufferedReader(new InputStreamReader(is));
   
   List<Instance<Double>> temp=new ArrayList<Instance<Double>>();
   String line=null;
   while((line=br.readLine())!=null)
   {
    String[] values = line.split(";");
    double label= Double.parseDouble(values[values.length-1]);
    double[] features=new double[values.length-1];
    for(int i=0; i < values.length-1; ++i)
    {
     features[i]=Double.parseDouble(values[i]);
    }
    
    Instance<Double> instance=new Instance<Double>(label, features);
    temp.add(instance);
   }
   
   Collections.shuffle(temp);
   
   for(Instance<Double> instance : temp)
   {
    if(testing_data.size() < 10)
    {
     testing_data.add(instance);
    }
    else
    {
     sample_data.add(instance);
    }
   }
   
  }catch(FileNotFoundException ex)
  {
   ex.printStackTrace();
  }catch(IOException ex)
  {
   ex.printStackTrace();
  }finally
  {
   try {
    if(is!=null) is.close();
    if(br !=null) br.close();
   } catch (IOException e) {
    e.printStackTrace();
   }
  }
 }
 
 public BirthDataSpout()
 {
  
 }
 
 public void open(Map conf, TopologyContext context) {
  // TODO Auto-generated method stub
  
 }

 public void emitBatch(long batchId, TridentCollector collector) {
  // TODO Auto-generated method stub
  int maxBatchCount=sample_data.size() / batchSize;
  if(maxBatchCount > 0 && batchIndex < maxBatchCount)
  {
   for(int i=batchIndex * batchSize; i < sample_data.size() && i < (batchIndex+1) * batchSize; ++i)
   {
    Instance<Double> instance = sample_data.get(i);
    collector.emit(new Values(instance));
   }
   batchIndex=(batchIndex+1) % maxBatchCount;
  }
 }

 public void ack(long batchId) {
  // TODO Auto-generated method stub
  
 }

 public void close() {
  // TODO Auto-generated method stub
  
 }

 public Map getComponentConfiguration() {
  // TODO Auto-generated method stub
  return null;
 }

 public Fields getOutputFields() {
  // TODO Auto-generated method stub
  return new Fields("instance");
 }

}

As can be seen above, the BirthDataSpout is derived from IBatchSpout, and emits a batch of 10 tuples at one time, each tuple is a training record containing the fields ("instance").  The "instance" field contains a datatype Instance<Double> which contains a double array as features and a double value as label. The training records are obtained from a births.csv file residing in src/test/resources.

PA Regression in Trident topology using Trident-ML implementation

Once we have the training data spout, we can build a Trident topology which uses the training data to create a predicted output value for each of the data record using PA regression algorithm in Trident-ML. This is implemented in the main class shown below:

package com.memeanalytics.trident_regression_pa;

import java.util.List;

import com.github.pmerienne.trident.ml.regression.PARegressor;
import com.github.pmerienne.trident.ml.regression.RegressionQuery;
import com.github.pmerienne.trident.ml.regression.RegressionUpdater;

import storm.trident.TridentState;
import storm.trident.TridentTopology;
import storm.trident.testing.MemoryMapState;
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.LocalDRPC;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.generated.StormTopology;
import backtype.storm.tuple.Fields;

/**
 * Hello world!
 *
 */
public class App 
{
    public static void main( String[] args ) throws AlreadyAliveException, InvalidTopologyException
    {
        LocalDRPC drpc=new LocalDRPC();
        
        LocalCluster cluster=new LocalCluster();
        Config config=new Config();
        
        cluster.submitTopology("RegressionDemo", config, buildTopology(drpc));
        
        try{
         Thread.sleep(10000);
        }catch(InterruptedException ex)
        {
         ex.printStackTrace();
        }
        
        List<String> drpc_args_list=BirthDataSpout.getDRPCArgsList();
        for(String drpc_args : drpc_args_list)
        {
         System.out.println(drpc.execute("predict", drpc_args));
        }
     
        cluster.killTopology("RegressionDemo");
        cluster.shutdown();
        
        drpc.shutdown();
    }
    
    private static StormTopology buildTopology(LocalDRPC drpc)
    {
     TridentTopology topology=new TridentTopology();
     
     BirthDataSpout spout=new BirthDataSpout();
     
     TridentState regressionModel = topology.newStream("training", spout).partitionPersist(new MemoryMapState.Factory(), new Fields("instance"), new RegressionUpdater("regression", new PARegressor()));
     
     topology.newDRPCStream("predict", drpc).each(new Fields("args"), new DRPCArgsToInstance(), new Fields("instance")).stateQuery(regressionModel, new Fields("instance"), new RegressionQuery("regression"), new Fields("prediction")).project(new Fields("args", "prediction"));
     
     return topology.build();
    }
}
package com.memeanalytics.trident_regression_pa;

import backtype.storm.tuple.Values;

import com.github.pmerienne.trident.ml.core.Instance;

import storm.trident.operation.BaseFunction;
import storm.trident.operation.TridentCollector;
import storm.trident.tuple.TridentTuple;

public class DRPCArgsToInstance extends BaseFunction {

 private static final long serialVersionUID = 1L;

 public void execute(TridentTuple tuple, TridentCollector collector) {
  String drpc_args = tuple.getString(0);
  String[] args=drpc_args.split(",");
  
  Double label=Double.parseDouble(args[args.length-1]);
  double[] features=new double[args.length-1];
  
  for(int i=0; i < args.length-1; ++i)
  {
   features[i]=Double.parseDouble(args[i]);
  }
  
  Instance<Double> instance=new Instance<Double>(label, features);
  
  collector.emit(new Values(instance));
 }

}

As can be seen above, the Trident topology has the BirthDataSpout emits Instance<Double> training data can be consumed by RegressionUpdater. The RegressionUpdater object from Trident-ML updates the underlying regressionModel via PA algorithm.

The DRPCStream allows user to pass in a new testing instance to the regressionModel which will then return a "predict" field, that contains the predicted output of the testing instance. The DRPCArgsToInstance is a BaseFunction operation which converts the arguments passed into the LocalDRPC.execute() into an Instance<Double> which can be passed into the RegressionQuery which then uses PARegressor and regressionModel to determine the predicted output value.

Once the coding is completed, we can run the project by navigating to the project root folder and run the following commands:

> .mvn compile exec:java

Trident-ML: Sentiment Analysis Classifier

Trident-ML comes with a pre-trained twitter sentiment classifier, this post shows how to use this classifier to perform sentiment analysis in Storm.

This post shows some very basic example of how to use the pre-trained twitter sentiment classifier in Trident-ML to classifier sentiment of text which will return true (positive) or false (negative).

Firstly create a Maven project (e.g. with groupId="com.memeanalytics" artifactId="trident-sentiment-classifier"). The complete source codes of the project can be downloaded from the link:

https://dl.dropboxusercontent.com/u/113201788/storm/trident-sentiment-classifier.tar.gz

For the start we need to configure the pom.xml file in the project.

Configure pom.xml:

Firstly we need to add the clojars repository to the repositories section:

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

Next we need to add the storm dependency to the dependencies section (for storm):

<dependency>
  <groupId>storm</groupId>
  <artifactId>storm</artifactId>
  <version>0.9.0.1</version>
  <scope>provided</scope>
</dependency>

Next we need to add the strident-ml dependency to the dependencies section (for text classification):

<dependency>
  <groupId>com.github.pmerienne</groupId>
  <artifactId>trident-ml</artifactId>
  <version>0.0.4</version>
</dependency>

Next we need to add the exec-maven-plugin to the build/plugins section (for execute the Maven project):

<plugin>
<groupId>org.codehaus.mojo</groupId>
<artifactId>exec-maven-plugin</artifactId>
<version>1.2.1</version>
<executions>
<execution>
<goals>
<goal>exec</goal>
</goals>
</execution>
</executions>
<configuration>
<includeProjectDependencies>true</includeProjectDependencies>
<includePluginDependencies>false</includePluginDependencies>
<executable>java</executable>
<classpathScope>compile</classpathScope>
<mainClass>com.memeanalytics.trident_sentiment_classifier.App</mainClass>
</configuration>
</plugin>

Next we need to add the maven-assembly-plugin to the build/plugins section (for packacging the Maven project to jar for submitting to Storm cluster):

<plugin>
<artifactId>maven-assembly-plugin</artifactId>
<version>2.2.1</version>
<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>

Sentiment Classification in Trident topology using Trident-ML implementation

Once the pom.xml update is completed, we can build a Trident topology which uses TwitterSentimentClassifier in a DRPCStream to classify text sentiment in Trident-ML. This is implemented in the main class shown below:

package com.memeanalytics.trident_sentiment_classifier;

import com.github.pmerienne.trident.ml.nlp.TwitterSentimentClassifier;

import storm.trident.TridentTopology;
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.LocalDRPC;
import backtype.storm.generated.StormTopology;
import backtype.storm.tuple.Fields;

public class App 
{
    public static void main( String[] args )
    {
        LocalDRPC drpc=new LocalDRPC();
        
        LocalCluster cluster=new LocalCluster();
        Config config=new Config();
        
        cluster.submitTopology("SentimentClassifierDemo", config, buildTopology(drpc));
        
        try{
         Thread.sleep(2000);
        }catch(InterruptedException ex)
        {
         ex.printStackTrace();
        }
        
        System.out.println(drpc.execute("classify", "Have a nice day!"));
        System.out.println(drpc.execute("classify", "I feel really bad!"));
        System.out.println(drpc.execute("classify", "Whatever, i don't really care"));
        System.out.println(drpc.execute("classify", "feel sleepy zzzz...."));
        
        cluster.killTopology("SentimentClassifierDemo");
        cluster.shutdown();
        drpc.shutdown();
    }
    
    private static StormTopology buildTopology(LocalDRPC drpc)
    {
     TridentTopology topology=new TridentTopology();
     
     topology.newDRPCStream("classify", drpc).each(new Fields("args"), new TwitterSentimentClassifier(), new Fields("sentiment"));
     
     return topology.build();
    }
}

The DRPCStream allows user to pass in a text string to the TwitterSentimentClassifier which will then return a "sentiment" field, that contains the predicted label (true for positive; false for negative) of the testing text.

Next copy the following two files into the "main/resources" folder under the project root folder:

twitter-sentiment-classifier-classifier.json:
https://github.com/pmerienne/trident-ml/blob/master/src/main/resources/twitter-sentiment-classifier-classifier.json

twitter-sentiment-classifier-extractor.json:
https://github.com/pmerienne/trident-ml/blob/master/src/main/resources/twitter-sentiment-classifier-extractor.json

The above step can be important, otherwise you may get a FileNotFoundException during runtime.

Once the coding is completed, we can run the project by navigating to the project root folder and run the following commands:

> .mvn compile exec:java

Trident-ML: Text Classification using KLD

This post shows some very basic example of how to use the Kullback-Leibler Distance text classification algorithm in Trident-ML to process data from Storm Spout.

Firstly create a Maven project (e.g. with groupId="com.memeanalytics" artifactId="trident-text-classifier-kld"). The complete source codes of the project can be downloaded from the link:

https://dl.dropboxusercontent.com/u/113201788/storm/trident-text-classifier-kld.tar.gz

For the start we need to configure the pom.xml file in the project.

Configure pom.xml:

Firstly we need to add the clojars repository to the repositories section:

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

Next we need to add the storm dependency to the dependencies section (for storm):

<dependency>
  <groupId>storm</groupId>
  <artifactId>storm</artifactId>
  <version>0.9.0.1</version>
  <scope>provided</scope>
</dependency>

Next we need to add the strident-ml dependency to the dependencies section (for text classification):

<dependency>
  <groupId>com.github.pmerienne</groupId>
  <artifactId>trident-ml</artifactId>
  <version>0.0.4</version>
</dependency>

Next we need to add the exec-maven-plugin to the build/plugins section (for execute the Maven project):

<plugin>
<groupId>org.codehaus.mojo</groupId>
<artifactId>exec-maven-plugin</artifactId>
<version>1.2.1</version>
<executions>
<execution>
<goals>
<goal>exec</goal>
</goals>
</execution>
</executions>
<configuration>
<includeProjectDependencies>true</includeProjectDependencies>
<includePluginDependencies>false</includePluginDependencies>
<executable>java</executable>
<classpathScope>compile</classpathScope>
<mainClass>com.memeanalytics.trident_text_classifier_kld.App</mainClass>
</configuration>
</plugin>

Next we need to add the maven-assembly-plugin to the build/plugins section (for packacging the Maven project to jar for submitting to Storm cluster):

<plugin>
<artifactId>maven-assembly-plugin</artifactId>
<version>2.2.1</version>
<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>

Implement Spout for training data 

Once the pom.xml update is completed, we can move to implement the ReuterNewsSpout which is the Storm spout that emits batches of training data to the Trident topology:

package com.memeanalytics.trident_text_classifier_kld;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import backtype.storm.task.TopologyContext;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
import storm.trident.operation.TridentCollector;
import storm.trident.spout.IBatchSpout;

public class ReuterNewsSpout implements IBatchSpout {

 private static final long serialVersionUID = 1L;
 private List<List<Object>> trainingData=new ArrayList<List<Object>>();
 private static Map<Integer, List<Object>> testingData=new HashMap<Integer, List<Object>>();
 
 private int batchSize=10;
 private int batchIndex=0;
 
 public ReuterNewsSpout()
 {
  try{
   loadReuterNews();
  }catch(FileNotFoundException ex)
  {
   ex.printStackTrace();
  }catch(IOException ex)
  {
   ex.printStackTrace();
  }
 }
 
 public static List<List<Object>> getTestingData()
 {
  List<List<Object>> result=new ArrayList<List<Object>>();
  for(Integer topic_index : testingData.keySet())
  {
   result.add(testingData.get(topic_index));
  }
  
  return result;
 }
 
 private void loadReuterNews() throws FileNotFoundException, IOException
 {
  Map<String, Integer> topics=new HashMap<String, Integer>();
  String filePath="src/test/resources/reuters.csv";
  FileInputStream inputStream=new FileInputStream(filePath);
  BufferedReader reader= new BufferedReader(new InputStreamReader(inputStream));
  String line;
  while((line = reader.readLine())!=null)
  {
   String topic = line.split(",")[0];
   if(!topics.containsKey(topic))
   {
    topics.put(topic, topics.size());
   }
   Integer topic_index=topics.get(topic);
   
   int index = line.indexOf(" - ");
   if(index==-1) continue;
   
   String text=line.substring(index, line.length()-1);
   
   if(testingData.containsKey(topic_index))
   {
    List<Object> values=new ArrayList<Object>();
    values.add(topic_index);
    values.add(text);
    trainingData.add(values);
   }
   else 
   {
    testingData.put(topic_index, new Values(topic_index, text));
   }
  }
  reader.close();
 }
 public void open(Map conf, TopologyContext context) {
  // TODO Auto-generated method stub
  
 }

 public void emitBatch(long batchId, TridentCollector collector) {
  // TODO Auto-generated method stub
  
  int maxBatchIndex = (trainingData.size() / batchSize);
  
  if(trainingData.size() > batchSize && batchIndex < maxBatchIndex)
  {
   for(int i=batchIndex * batchSize; i < trainingData.size() && i < (batchIndex+1) * batchSize; ++i)
   {
    collector.emit(trainingData.get(i));
   }
   
   
   batchIndex++;
   
   //System.out.println("Progress: "+batchIndex +" / "+maxBatchIndex);
  }
 }

 public void ack(long batchId) {
  // TODO Auto-generated method stub
  
 }

 public void close() {
  // TODO Auto-generated method stub
  
 }

 public Map getComponentConfiguration() {
  // TODO Auto-generated method stub
  return null;
 }

 public Fields getOutputFields() {
  // TODO Auto-generated method stub
  return new Fields("label", "text");
 }

}


As can be seen above, the ReuterNewsSpout is derived from IBatchSpout, and emits a batch of 10 tuples at one time, each tuple is a new article containing the fields ("label", "text"). The "label" field is integer value (represents the topic of the news article), while "text" field is a string which is text of the news article. the training records are obtained in such a way that the correct prediction learned from the text classification should be predicting the topic of a news article given the text of the news article.

KLD Text Classification in Trident topology using Trident-ML implementation

Once we have the training data spout, we can build a Trident topology which uses the training data to create a class label for each of the data record using KLD classifier algorithm in Trident-ML. This is implemented in the main class shown below:

package com.memeanalytics.trident_text_classifier_kld;

import java.util.List;

import com.github.pmerienne.trident.ml.nlp.ClassifyTextQuery;
import com.github.pmerienne.trident.ml.nlp.KLDClassifier;
import com.github.pmerienne.trident.ml.nlp.TextClassifierUpdater;
import com.github.pmerienne.trident.ml.preprocessing.TextInstanceCreator;

import storm.trident.TridentState;
import storm.trident.TridentTopology;
import storm.trident.testing.MemoryMapState;
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.LocalDRPC;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.generated.StormTopology;
import backtype.storm.tuple.Fields;


public class App 
{
    public static void main( String[] args ) throws AlreadyAliveException, InvalidTopologyException
    {
        LocalDRPC drpc=new LocalDRPC();
        
        LocalCluster cluster=new LocalCluster();
        
        Config config=new Config();
        
        cluster.submitTopology("KLDDemo", config, buildTopology(drpc));
        
        try{
         Thread.sleep(20000);
        }catch(InterruptedException ex)
        {
         ex.printStackTrace();
        }
        
        List<List<Object>> testingData = ReuterNewsSpout.getTestingData();
        
        for(int i=0; i < testingData.size(); ++i)
        {
         List<Object> testingDataRecord=testingData.get(i);
         String drpc_args="";
         for(Object val : testingDataRecord){
          if(drpc_args.equals(""))
          {
           drpc_args+=val;
          }
          else
          {
           drpc_args+=(","+val);
          }
         }
         System.out.println(drpc.execute("predict", drpc_args));
        }
        
        cluster.killTopology("KLDDemo");
        cluster.shutdown();
        
        drpc.shutdown();
    }
    
    private static StormTopology buildTopology(LocalDRPC drpc)
    {
     ReuterNewsSpout spout=new ReuterNewsSpout();
     
     TridentTopology topology=new TridentTopology();
     
     TridentState classifierModel = topology.newStream("training", spout).each(new Fields("label", "text"), new TextInstanceCreator<Integer>(), new Fields("instance")).partitionPersist(new MemoryMapState.Factory(), new Fields("instance"), new TextClassifierUpdater("newsClassifier", new KLDClassifier(9)));
     
     topology.newDRPCStream("predict", drpc).each(new Fields("args"), new DRPCArgsToInstance(), new Fields("instance")).stateQuery(classifierModel, new Fields("instance"), new ClassifyTextQuery("newsClassifier"), new Fields("prediction"));
     return topology.build();
    }
}

package com.memeanalytics.trident_text_classifier_kld;

import java.util.ArrayList;
import java.util.List;

import backtype.storm.tuple.Values;

import com.github.pmerienne.trident.ml.core.TextInstance;
import com.github.pmerienne.trident.ml.preprocessing.EnglishTokenizer;
import com.github.pmerienne.trident.ml.preprocessing.TextTokenizer;

import storm.trident.operation.BaseFunction;
import storm.trident.operation.TridentCollector;
import storm.trident.tuple.TridentTuple;

public class DRPCArgsToInstance extends BaseFunction{

 private static final long serialVersionUID = 1L;

 public void execute(TridentTuple tuple, TridentCollector collector) {
  // TODO Auto-generated method stub
  String drpc_args=tuple.getString(0);
  String[] args=drpc_args.split(",");
  Integer label=Integer.parseInt(args[0]);
  
  String text=args[1];
  
  TextTokenizer textAnalyzer=new EnglishTokenizer();
  List<String> tokens=textAnalyzer.tokenize(text);
  
  
  TextInstance<Integer> instance=new TextInstance<Integer>(label, tokens);
  
  collector.emit(new Values(instance));
 }

}

As can be seen above, the Trident topology has a TextInstanceCreator<Integer> trident operation which convert raw ("label", "text") tuple into an TextInstance<Integer> object which can be consumed by TextClassifierUpdater. The TextClassifierUpdater object from Trident-ML updates the underlying classifierModel via KLDClassifier training algorithm.

The DRPCStream allows user to pass in a new testing instance to the classifierModel which will then return a "predict" field, that contains the predicted label of the testing instance. The DRPCArgsToInstance is a BaseFunction operation which converts the arguments passed into the LocalDRPC.execute() into an TextInstance<Integer> (Note you can set the label to null in DRPCArgsToInstance.execute() method as the label will be predicted instead) which can be passed into the ClassifyTextQuery which then uses KLD and classifierModel to determine the predicted label.

Once the coding is completed, we can run the project by navigating to the project root folder and run the following commands:

> .mvn compile exec:java

Trident-ML: Classification using Perceptron

This post shows some very basic example of how to use the perceptron classification algorithm in Trident-ML to process data from Storm Spout.

Firstly create a Maven project (e.g. with groupId="com.memeanalytics" artifactId="trident-classifier-perceptron"). The complete source codes of the project can be downloaded from the link:

https://dl.dropboxusercontent.com/u/113201788/storm/trident-classifier-perceptron.tar.gz

For the start we need to configure the pom.xml file in the project.

Configure pom.xml:
Firstly we need to add the clojars repository to the repositories section:

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

Next we need to add the storm dependency to the dependencies section (for storm):

<dependency>
  <groupId>storm</groupId>
  <artifactId>storm</artifactId>
  <version>0.9.0.1</version>
  <scope>provided</scope>
</dependency>

Next we need to add the strident-ml dependency to the dependencies section (for perceptron classification):

<dependency>
  <groupId>com.github.pmerienne</groupId>
  <artifactId>trident-ml</artifactId>
  <version>0.0.4</version>
</dependency>

Next we need to add the exec-maven-plugin to the build/plugins section (for execute the Maven project):

<plugin>
<groupId>org.codehaus.mojo</groupId>
<artifactId>exec-maven-plugin</artifactId>
<version>1.2.1</version>
<executions>
<execution>
<goals>
<goal>exec</goal>
</goals>
</execution>
</executions>
<configuration>
<includeProjectDependencies>true</includeProjectDependencies>
<includePluginDependencies>false</includePluginDependencies>
<executable>java</executable>
<classpathScope>compile</classpathScope>
<mainClass>com.memeanalytics.trident_classifier_perceptron.App</mainClass>
</configuration>
</plugin>

Next we need to add the maven-assembly-plugin to the build/plugins section (for packacging the Maven project to jar for submitting to Storm cluster):

<plugin>
<artifactId>maven-assembly-plugin</artifactId>
<version>2.2.1</version>
<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>

Implement Spout for training data 

Once the pom.xml update is completed, we can move to implement the NANDSpout which is the Storm spout that emits batches of training data to the Trident topology:

package com.memeanalytics.trident_classifier_perceptron;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Random;

import backtype.storm.task.TopologyContext;
import backtype.storm.tuple.Fields;
import storm.trident.operation.TridentCollector;
import storm.trident.spout.IBatchSpout;

public class NANDSpout implements IBatchSpout {

 private int batchSize=10;
 
 public void open(Map conf, TopologyContext context) {
  // TODO Auto-generated method stub
  
 }

 public void emitBatch(long batchId, TridentCollector collector) {
  // TODO Auto-generated method stub
  final Random rand=new Random();
  for(int i=0; i < batchSize; ++i)
  {
   boolean x0=rand.nextBoolean();
   boolean x1=rand.nextBoolean();
   boolean label = !(x0 && x1);
   List<Object> values=new ArrayList<Object>();
   values.add(label);
   values.add(x0 ? 1.0 : 0.0);
   values.add(x1 ? 1.0 : 0.0);
   //values.add(x0 ? 1.0 + noise(rand) : 0.0 + noise(rand));
   //values.add(x1 ? 1.0 + noise(rand) : 0.0 + noise(rand));
   collector.emit(values);
  }
 }
 
 public static double noise(Random rand)
 {
  return rand.nextDouble()* 0.0001 - 0.00005;
 }

 public void ack(long batchId) {
  // TODO Auto-generated method stub
  
 }

 public void close() {
  // TODO Auto-generated method stub
  
 }

 public Map getComponentConfiguration() {
  // TODO Auto-generated method stub
  return null;
 }

 public Fields getOutputFields() {
  // TODO Auto-generated method stub
  return new Fields("label", "x0", "x1");
 }

}

As can be seen above, the NANDSpout is derived from IBatchSpout, and emits a batch of 10 tuples at one time, each tuple is a training record containing the fields ("label", "x0", "x1"). The label is boolean value, while x0, x1 are double values which are either 1 (true) or 0 (false). the training records are obtained in such a way that the correct prediction should be a NAND gate from the classification.

Perceptron Classification in Trident topology using Trident-ML implementation

Once we have the training data spout, we can build a Trident topology which uses the training data to create a class label for each of the data record using perceptron classifier algorithm in Trident-ML. This is implemented in the main class shown below:

package com.memeanalytics.trident_classifier_perceptron;

import java.util.Random;

import com.github.pmerienne.trident.ml.classification.ClassifierUpdater;
import com.github.pmerienne.trident.ml.classification.ClassifyQuery;
import com.github.pmerienne.trident.ml.classification.PerceptronClassifier;
import com.github.pmerienne.trident.ml.preprocessing.InstanceCreator;

import storm.trident.TridentState;
import storm.trident.TridentTopology;
import storm.trident.testing.MemoryMapState;
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.LocalDRPC;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.generated.StormTopology;
import backtype.storm.tuple.Fields;

/**
 * Hello world!
 *
 */
public class App 
{
    public static void main( String[] args ) throws AlreadyAliveException, InvalidTopologyException
    {
        LocalDRPC drpc=new LocalDRPC();
        
        LocalCluster cluster=new LocalCluster();
        Config config=new Config();
        
        cluster.submitTopology("PerceptronDemo", config, buildTopology(drpc));
        
        try{
         Thread.sleep(10000);
        }catch(InterruptedException ex)
        {
         ex.printStackTrace();
        }
        
        for(int i=0; i < 10; ++i)
        {
         String drpc_args=createDRPCTestingSample();
         System.out.println(drpc.execute("predict", drpc_args));
         try{
          Thread.sleep(1000);
         }catch(InterruptedException ex)
         {
          ex.printStackTrace();
         }
        }
        
        cluster.killTopology("PerceptronDemo");
        cluster.shutdown();
        
        drpc.shutdown();
    }
    
    private static String createDRPCTestingSample()
    {
     String drpc_args="";
     
     final Random rand=new Random();
     
     boolean bit_x0=rand.nextBoolean();
  boolean bit_x1=rand.nextBoolean();
  boolean label = !(bit_x0 && bit_x1);
  
  double x0=bit_x0 ? 1.0 + NANDSpout.noise(rand) : 0.0 + NANDSpout.noise(rand);
  double x1=bit_x1 ? 1.0 + NANDSpout.noise(rand) : 0.0 + NANDSpout.noise(rand);
  
  drpc_args+=label;
  drpc_args+=(","+x0);
  drpc_args+=(","+x1);
  
  return drpc_args;
    }
    
    private static StormTopology buildTopology(LocalDRPC drpc)
    {
     TridentTopology topology=new TridentTopology();
     NANDSpout spout=new NANDSpout();
     TridentState classifierModel = topology.newStream("training", spout).shuffle().each(new Fields("label", "x0", "x1"), new InstanceCreator<Boolean>(), new Fields("instance")).partitionPersist(new MemoryMapState.Factory(), new Fields("instance"), new ClassifierUpdater<Boolean>("perceptron", new PerceptronClassifier()));
     
     topology.newDRPCStream("predict", drpc).each(new Fields("args"), new DRPCArgsToInstance(), new Fields("instance")).stateQuery(classifierModel, new Fields("instance"), new ClassifyQuery<Boolean>("perceptron"), new Fields("predict"));
     
     return topology.build();
     
    }
}
package com.memeanalytics.trident_classifier_perceptron;

import backtype.storm.tuple.Values;

import com.github.pmerienne.trident.ml.core.Instance;

import storm.trident.operation.BaseFunction;
import storm.trident.operation.TridentCollector;
import storm.trident.tuple.TridentTuple;

public class DRPCArgsToInstance extends BaseFunction {

 private static final long serialVersionUID = 1L;

 public void execute(TridentTuple tuple, TridentCollector collector) {
  // TODO Auto-generated method stub
  String drpc_args=tuple.getString(0);
  String[] args=drpc_args.split(",");
  boolean label=Boolean.parseBoolean(args[0]);
  
  double[] features=new double[args.length-1];
  for(int i=1; i < args.length; ++i)
  {
   features[i-1]=Double.parseDouble(args[i]);
  }
  
  Instance<Boolean> instance=new Instance<Boolean>(label, features);
  
  collector.emit(new Values(instance));
 }

}

As can be seen above, the Trident topology has a InstanceCreator<Boolean> trident operation which convert raw ("label", "x0", "x1") tuple into an Instance<Boolean> object which can be consumed by ClassifierUpdater. The ClassifierUpdater object from Trident-ML updates the underlying classifierModel via perceptron training algorithm.

The DRPCStream allows user to pass in a new testing instance to the classifierModel which will then return a "predict" field, that contains the predicted label of the testing instance. The DRPCArgsToInstance is a BaseFunction operation which converts the arguments passed into the LocalDRPC.execute() into an Instance<Boolean> which can be passed into the ClassifyQuery which then uses perceptron and classifierModel to determine the predicted label.

Once the coding is completed, we can run the project by navigating to the project root folder and run the following commands:

> .mvn compile exec:java

Monday, November 24, 2014

Trident-ML: Clustering using K-Means

This post shows some very basic example of how to use the k means clustering algorithm in Trident-ML to process data from Storm Spout.

Firstly create a Maven project (e.g. with groupId="com.memeanalytics" artifactId="trident-k-means"). The complete source codes of the project can be downloaded from the link:

https://dl.dropboxusercontent.com/u/113201788/storm/trident-k-means.tar.gz

For the start we need to configure the pom.xml file in the project.

Configure pom.xml:
Firstly we need to add the clojars repository to the repositories section:

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

Next we need to add the storm dependency to the dependencies section (for storm):

<dependency>
  <groupId>storm</groupId>
  <artifactId>storm</artifactId>
  <version>0.9.0.1</version>
  <scope>provided</scope>
</dependency>

Next we need to add the strident-ml dependency to the dependencies section (for k-means clustering):

<dependency>
  <groupId>com.github.pmerienne</groupId>
  <artifactId>trident-ml</artifactId>
  <version>0.0.4</version>
</dependency>

Next we need to add the exec-maven-plugin to the build/plugins section (for execute the Maven project):

<plugin>
<groupId>org.codehaus.mojo</groupId>
<artifactId>exec-maven-plugin</artifactId>
<version>1.2.1</version>
<executions>
<execution>
<goals>
<goal>exec</goal>
</goals>
</execution>
</executions>
<configuration>
<includeProjectDependencies>true</includeProjectDependencies>
<includePluginDependencies>false</includePluginDependencies>
<executable>java</executable>
<classpathScope>compile</classpathScope>
<mainClass>com.memeanalytics.trident_k_means.App</mainClass>
</configuration>
</plugin>

Next we need to add the maven-assembly-plugin to the build/plugins section (for packacging the Maven project to jar for submitting to Storm cluster):

<plugin>
<artifactId>maven-assembly-plugin</artifactId>
<version>2.2.1</version>
<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>

Implement Spout for training data 

Once the pom.xml update is completed, we can move to implement the RandomFeatureSpout which is the Storm spout that emits batches of training data to the Trident topology:

package com.memeanalytics.trident_k_means;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import com.github.pmerienne.trident.ml.core.Instance;
import com.github.pmerienne.trident.ml.testing.data.Datasets;

import backtype.storm.task.TopologyContext;
import backtype.storm.tuple.Fields;
import storm.trident.operation.TridentCollector;
import storm.trident.spout.IBatchSpout;

public class RandomFeatureSpout implements IBatchSpout{

 private int batchSize=10;
 private int numFeatures=3;
 private int numClasses=3;
 
 public void open(Map conf, TopologyContext context) {
  // TODO Auto-generated method stub
  
 }

 public void emitBatch(long batchId, TridentCollector collector) {
  // TODO Auto-generated method stub
  List<Instance<Integer>> data = Datasets.generateDataForMultiLabelClassification(batchSize, numFeatures, numClasses);
  
  
  for(Instance<Integer> instance : data)
  {
   List<Object> values=new ArrayList<Object>();
   values.add(instance.label);
   
   for(double feature : instance.getFeatures())
   {
    values.add(feature);
   }
   collector.emit(values);
  }
  
 }

 public void ack(long batchId) {
  // TODO Auto-generated method stub
  
 }

 public void close() {
  // TODO Auto-generated method stub
  
 }

 public Map getComponentConfiguration() {
  // TODO Auto-generated method stub
  return null;
 }

 public Fields getOutputFields() {
  // TODO Auto-generated method stub
  return new Fields("label", "x0", "x1", "x2");
 }
}

As can be seen above, the RandomFeatureSpout is derived from IBatchSpout, and emits a batch of 10 tuples at one time, each tuple is a training record containing the fields ("label", "x0", "x1", "x2"). The label is integer, while x0, x1, x2 are double values. the training records are obtained from Trident-ML's DataSets.generateDataForMultiLabelClassification() method.

K-means in Trident topology using Trident-ML implementation

Once we have the training data spout, we can build a Trident topology which uses the training data to create a class label for each of the data record using k-means algorithm in Trident-ML. This is implemented in the main class shown below:

package com.memeanalytics.trident_k_means;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;

import com.github.pmerienne.trident.ml.clustering.ClusterQuery;
import com.github.pmerienne.trident.ml.clustering.ClusterUpdater;
import com.github.pmerienne.trident.ml.clustering.KMeans;
import com.github.pmerienne.trident.ml.core.Instance;
import com.github.pmerienne.trident.ml.preprocessing.InstanceCreator;
import com.github.pmerienne.trident.ml.testing.data.Datasets;

import storm.trident.TridentState;
import storm.trident.TridentTopology;
import storm.trident.testing.MemoryMapState;
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.LocalDRPC;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.generated.StormTopology;
import backtype.storm.tuple.Fields;

public class App 
{
    public static void main( String[] args ) throws AlreadyAliveException, InvalidTopologyException
    {
        LocalDRPC drpc=new LocalDRPC();
        
        Config config=new Config();
        
        LocalCluster cluster=new LocalCluster();
        
        cluster.submitTopology("KMeansDemo", config, buildTopology(drpc));
        
        try{
         Thread.sleep(10000);
        }catch(InterruptedException ex)
        {
         ex.printStackTrace();
        }
        
        
        
        for(int i=0; i < 10; ++i)
        {
         String drpc_args=generateRandomTestingArgs();
         System.out.println(drpc.execute("predict", drpc_args));
         try{
          Thread.sleep(1000);
         }catch(InterruptedException ex)
         {
          ex.printStackTrace();
         }
        }
        
        cluster.killTopology("KMeansDemo");
        cluster.shutdown();
        drpc.shutdown();
    }
    
    private static String generateRandomTestingArgs()
    {
     int batchSize=10;
     int numFeatures=3;
     int numClasses=3;
     
     final Random rand=new Random();
     
     List<Instance<Integer>> data = Datasets.generateDataForMultiLabelClassification(batchSize, numFeatures, numClasses);
  
  String args="";
  Instance<Integer> instance = data.get(rand.nextInt(data.size()));
  
  
  args+=instance.label;
  
  for(double feature : instance.getFeatures())
  {
   args+=(","+feature);
  }
   
  
  return args;
    }
    
    private static StormTopology buildTopology(LocalDRPC drpc)
    {
     TridentTopology topology=new TridentTopology();
     
     RandomFeatureSpout spout=new RandomFeatureSpout();
     
     TridentState clusterModel = topology.newStream("training", spout).each(new Fields("label", "x0", "x1", "x2"), new InstanceCreator<Integer>(), new Fields("instance")).partitionPersist(new MemoryMapState.Factory(), new Fields("instance"), new ClusterUpdater("kmeans", new KMeans(3)));
     
     topology.newDRPCStream("predict", drpc).each(new Fields("args"), new DRPCArgsToInstance(), new Fields("instance")).stateQuery(clusterModel, new Fields("instance"), new ClusterQuery("kmeans"), new Fields("predict"));
     
     return topology.build();
    }
}
package com.memeanalytics.trident_k_means;

import java.util.ArrayList;
import java.util.List;

import backtype.storm.tuple.Values;

import com.github.pmerienne.trident.ml.core.Instance;

import storm.trident.operation.BaseFunction;
import storm.trident.operation.TridentCollector;
import storm.trident.tuple.TridentTuple;

public class DRPCArgsToInstance extends BaseFunction{

 private static final long serialVersionUID = 1L;

 public void execute(TridentTuple tuple, TridentCollector collector) {
  // TODO Auto-generated method stub
  String drpc_args = tuple.getString(0);
  String[] args = drpc_args.split(",");
  Integer label=Integer.parseInt(args[0]);
  double[] features=new double[args.length-1];
  for(int i=1; i < args.length; ++i)
  {
   double feature=Double.parseDouble(args[i]);
   features[i-1] = feature;
  }
  Instance<Integer> instance=new Instance<Integer>(label, features);
  
  collector.emit(new Values(instance));
 }

}

As can be seen above, the Trident topology has a InstanceCreator<Integer> trident operation which convert raw ("label", "x0", "x1", "x2") tuple into an Instance<Integer> object which can be consumed by ClusterUpdator. The ClusterUpdate object from Trident-ML updates the underlying clusterModel via k-Means algorithm.

The DRPCStream allows user to pass in a new testing instance to the clusterModel which will then return a "predict" field, that contains the predicted label of the testing instance. The DRPCArgsToInstance is a BaseFunction operation which converts the arguments passed into the LocalDRPC.execute() into an Instance<Integer> which can be passed into the ClusterQuery which then uses kmeans and clusterModel to determine the predicted label.

Once the coding is completed, we can run the project by navigating to the project root folder and run the following commands:

> .mvn compile exec:java

Integration of Kafka-Trident-MySQL

This post shows a most basic example in which user can integrate Kafka, Trident (on top of Storm) and MySQL. The example uses a Kafka producer which randomly produce messages to Kafka brokers (a random list of country names), a TransactionalTridentKafkaSpout is used pull data from Kafka messaging system and emits the tuples (containing the field "str" which is the country names from the Kafka producer) to a Trident operation that serialize the received messages into the mysql database.

Some ZooKeeper and Kafka settings need to be explained before we proceed. the source codes developed here assumes that the ZooKeepers runs on the following nodes:

192.168.2.2:2181
192.168.2.4:2181

and also assumes that the Kafka brokers runs at the following hostname:port:

192.168.2.2:9092
192.168.2.4:9092

The Kafka producer can be downloaded from the following link:

https://dl.dropboxusercontent.com/u/113201788/storm/kakfa-producer-for-trident.tar.gz

Basically the Kafka producer emits a random list of country names as messages sent to the Kafka brokers. The tutorial on how to implement Kafka producer can be found at:

Next we need to create Maven project (e.g. with groupId="com.memeanalytics" and artifactId="kafka-trident-consumer") which will consumes the Kafka message in a Trident topology and serialize it to mysql database. The source codes of the project can be downloaded from:

https://dl.dropboxusercontent.com/u/113201788/storm/kafka-trident-consumer.zip

Below we will explain how to prepare the pom.xml file, implement the storm operation for mysql serialization, as well as configuration of Trident topology which can consume messages from Kafka brokers.

Prepare pom.xml

In the pom.xml, first add in the clojars repository in the repositories section:

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

Next add in the storm-kafka-0.8-plus dependency in the dependencies section (for TransactionalTridentKafkaSpout):

<dependency>
  <groupId>net.wurstmeister.storm</groupId>
  <artifactId>storm-kafka-0.8-plus</artifactId>
  <version>0.4.0</version>
</dependency>

Next add in the storm-core dependency in the dependencies section (for storm):

<dependency>
  <groupId>storm</groupId>
  <artifactId>storm-core</artifactId>
  <version>0.9.0.1</version>
</dependency>

Next add in the mysql-connector-java dependency in the dependencies section (for mysql):

<dependency>
  <groupId>mysql</groupId>
  <artifactId>mysql-connector-java</artifactId>
  <version>5.1.6</version>
</dependency>

Next add in the maven-assembly-plugin in the build/plugins section (for packaging the Maven project as jar for submitting to Storm cluster):

<plugin>
<artifactId>maven-assembly-plugin</artifactId>
<version>2.2.1</version>
<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>

Next add in the exec-maven-plugin in the build/plugins section (for executing the Maven project):

<plugin>
<groupId>org.codehaus.mojo</groupId>
<artifactId>exec-maven-plugin</artifactId>
<version>1.2.1</version>

<executions>
<execution>
<goals>
<goal>exec</goal>
</goals>
</execution>
</executions>

<configuration>
<includeProjectDependencies>true</includeProjectDependencies>
<includePluginDependencies>false</includePluginDependencies>
<executable>java</executable>
<classpathScope>compile</classpathScope>
<mainClass>com.memeanalytics.kafka_trident_consumer.App</mainClass>
</configuration>

</plugin>

This completes the pom.xml configuration. Next we will implements the Trident operation which serializes the TridentTuple data into mysql database.

Trident operation for mysql serialization

The Trident operation which serializes the TridentTuple data is a BaseFilter object from Trident that has the following implementation:

package com.memeanalytics.kafka_trident_consumer;

import java.util.Map;

import storm.trident.operation.BaseFilter;
import storm.trident.operation.TridentOperationContext;
import storm.trident.tuple.TridentTuple;

public class TridentUtils {
 public static class MySqlPersist extends BaseFilter{

  private static final long serialVersionUID = 1L;

  private MySqlDump mysqlSerializer=null;
  
  public boolean isKeep(TridentTuple tuple) {
   String country=tuple.getString(0);
   mysqlSerializer.store(country);
   System.out.println("Country: "+country);
   return true;
  }
  
  @Override
     public void prepare(Map conf, TridentOperationContext context) {
   mysqlSerializer=new MySqlDump("localhost","mylog", "root", "[username]");
     }

     @Override
     public void cleanup() {
      mysqlSerializer.close();
     }
  
 }
}

In the above implementation, the mysqlSerializer member variable is responsible for actually storing the data in mysql database. it opens the mysql connection (in its constructor) in prepare() method and closes the mysql connection in cleanup() method. the data serialization happens in the execute() method. Below is the implementation of the class of the variable:

package com.memeanalytics.kafka_trident_consumer;

import java.sql.Connection;
import java.sql.SQLException;

import java.sql.PreparedStatement;

public class MySqlDump {
 private String ip;
 private String database;
 private String username;
 private String password;
 private Connection conn;
 
 public MySqlDump(String ip, String database, String username, String password)
 {
  this.ip = ip;
  this.database=database;
  this.username=username;
  this.password=password;
  conn=MySqlConnectionGenerator.open(ip, database, username, password);
 }
 
 public void store(String dataitem)
 {
  if(conn==null) return;
  
  PreparedStatement statement = null;
  try{
   statement = conn.prepareStatement("insert into mylogtable (id, dataitem) values (default, ?)");
   statement.setString(1, dataitem);
   statement.executeUpdate();
  }catch(Exception ex)
  {
   ex.printStackTrace();
  }finally
  {
   if(statement != null)
   {
    try {
     statement.close();
    } catch (SQLException e) {
     // TODO Auto-generated catch block
     e.printStackTrace();
    }
   }
  }
 }
 
 public void close()
 {
  if(conn==null) return;
  try {
   conn.close();
  } catch (SQLException e) {
   // TODO Auto-generated catch block
   e.printStackTrace();
  }
 }
}
package com.memeanalytics.kafka_trident_consumer;

import java.sql.Connection;
import java.sql.DriverManager;

public class MySqlConnectionGenerator {
 public static Connection open(String ip, String database, String username, String password)
 {
  Connection conn = null;
  try{
   Class.forName("com.mysql.jdbc.Driver");
   conn=DriverManager.getConnection("jdbc:mysql://"+ip+"/"+database+"?user="+username+"&password="+password);
  }catch(Exception ex)
  {
   ex.printStackTrace();
  }
  
  return conn;
 }
}

Trident topology implementation

Once this is completed. we are ready to implement the Trident topology which consumes data from Kafka and saves it to mysql. this is implemented in the main class:

package com.memeanalytics.kafka_trident_consumer;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.spout.SchemeAsMultiScheme;
import backtype.storm.tuple.Fields;
import storm.kafka.SpoutConfig;
import storm.kafka.StringScheme;
import storm.kafka.ZkHosts;
import storm.kafka.trident.TransactionalTridentKafkaSpout;
import storm.kafka.trident.TridentKafkaConfig;
import storm.trident.TridentTopology;

public class App 
{
    public static void main( String[] args ) throws AlreadyAliveException, InvalidTopologyException
    {
        ZkHosts zkHosts=new ZkHosts("192.168.2.4:2181");
        
        String topic="country-topic";
        String consumer_group_id="storm";
        
        TridentKafkaConfig kafkaConfig=new TridentKafkaConfig(zkHosts, topic, consumer_group_id);
        
        kafkaConfig.scheme=new SchemeAsMultiScheme(new StringScheme());
        kafkaConfig.forceFromStart=true;
        
        TransactionalTridentKafkaSpout spout=new TransactionalTridentKafkaSpout(kafkaConfig);
        
        TridentTopology topology=new TridentTopology();
        
        topology.newStream("spout", spout).shuffle().each(new Fields("str"), new TridentUtils.MySqlPersist());
        
        LocalCluster cluster=new LocalCluster();
        
        Config config=new Config();
        
        cluster.submitTopology("KafkaTridentMysqlDemo", config, topology.build());
        
        try{
         Thread.sleep(10000);
        }catch(InterruptedException ex)
        {
         ex.printStackTrace();
        }
        
        cluster.killTopology("KafkaTridentMysqlDemo");
        cluster.shutdown();
    }
}

For demo purpose, its uses LocalCluster to submit and run the Trident topology and does not implement a DRPCStream which can be used to queried result. The main change is the TractionalTridentKafkaSpout which takes in a TridentKafkaConfig object as parameter in its constructor. The TransactionalTridentKafkaSpout emits tuples which is serialized by the TridentUtil.MySqlPersist that is the mysql serialization Trident operation.

Once done, we can compile and exec the Maven project by navigating to its root folder and run the following command:

> mvn compile exec:java

Integration of Kafka-Storm-MySQL

This post will discuss how to create a minimum basic storm topology which integrate Kafka, Storm and MySQL. The scenerios is: a Kafka producer will push some dummy data to the Kafka brokers, and a KafkaSpout (with consumer group id = "id7" and zookeeper = "192.168.2.4:2181") from storm cluster consumes data from the Kafka brokers. The KafkaSpout will then emits Kafka messages as tuples to a BaseBasicBolt which will then persists the data to the MySQL server.

The post assumes user already have a zookeeper cluster set up on two hostname:port:

192.168.2.2:2181
192.168.2.4:2181

The post also assumes that the Kafka brokers runs at the following hostname:port:

192.168.2.2:9092
192.168.2.4:9092

Details of how to set up ZooKeeper and Kafka cluster can be found at the following links:

http://czcodezone.blogspot.sg/2014/11/setup-zookeeper-in-cluster.html
http://czcodezone.blogspot.sg/2014/11/setup-kafka-in-cluster.html

The Kafka producer source codes can be downloaded from the link: https://dl.dropboxusercontent.com/u/113201788/kafka-producer.zip. The details about how to implement a simple Kafka producer in java can be found at this link: http://czcodezone.blogspot.sg/2014/11/write-and-test-simple-kafka-producer.html).

To create the Storm topology which consumes messages from Kafka and persists them to MySQL database, create a Maven project (e.g. with groupId="com.memeanalytics" and artifactId="storm-kafka-mysql"). The complete source code of the project can be downloaded from the following link:

https://dl.dropboxusercontent.com/u/113201788/storm/storm-kafka-mysql.tar.gz

Maven setup: pom.xml

We will start by explaining setup in the pom.xml file. Firstly, we need to put in the clojars in the repositories:

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

Next we add the mysql dependency into the dependencies section:

<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>5.1.6</version>
</dependency>

Next we add the storm-kafka-0.8-plus dependency into the dependencies section (for KafkaSpout which consumers Kafka messages in Storm):

<dependency>
<groupId>net.wurstmeister.storm</groupId>
<artifactId>storm-kafka-0.8-plus</artifactId>
<version>0.4.0</version>
</dependency>

Next we add the storm-core dependency into the dependencies section (for Storm):

<dependency>
<groupId>storm</groupId>
<artifactId>storm-core</artifactId>
<version>0.9.0.1</version>
<scope>provided</scope>
</dependency>

Next we add the exec-maven-plugin into the build/plugins section (for compile and execute java project):

<plugin>
<groupId>org.codehaus.mojo</groupId>
<artifactId>exec-maven-plugin</artifactId>
<version>1.2.1</version>
<executions>
<execution>
<goals>
<goal>exec</goal>
</goals>
</execution>
</executions>
<configuration>
<includeProjectDependencies>true</includeProjectDependencies>
<includePluginDependencies>false</includePluginDependencies>
<executable>java</executable>
<classpathScope>compile</classpathScope>
<mainClass>${main.class}</mainClass>
</configuration>
</plugin>

Next we add the maven-assembly-plugin into the build/plugins section (for packaging the java project as jar to submit to the Storm cluster):

<plugin>
<artifactId>maven-assembly-plugin</artifactId>
<version>2.2.1</version>
<configuration>
<descriptorRefs>
<descriptorRef>jar-with-dependencies</descrptorRef>
</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>

MySQL Connection and Management Classes

Next we will create a class that manages MySQL connection, the source codes of which are show below:

package com.memeanalytics.storm_kafka_mysql;

import java.sql.Connection;
import java.sql.DriverManager;


public class MySqlConnection {
 private String ip;
 private String database;
 private String username;
 private String password;
 private Connection conn;
 
 public MySqlConnection(String ip, String database, String username, String password)
 {
  this.ip=ip;
  this.database=database;
  this.username=username;
  this.password=password;
 }
 
 public Connection getConnection()
 {
  return conn;
 }
 
 public boolean open()
 {
  boolean successful=true;
  try{
   Class.forName("com.mysql.jdbc.Driver");
   conn = DriverManager.getConnection("jdbc:mysql://"+ip+"/"+database+"?"+"user="+username+"&password="+password);
  }catch(Exception ex)
  {
   successful=false;
   ex.printStackTrace();
  }
  return successful;
 }
 
 public boolean close()
 {
  if(conn==null)
  {
   return false;
  }
  
  boolean successful=true;
  try{
   conn.close();
  }catch(Exception ex)
  {
   successful=false;
   ex.printStackTrace();
  }
  
  return successful;
 }
}

As illustrated above, the MySqlConnection manages the opening and closing of the connection.

Next we will create a class which saves the tuples emitted from the KafkaSpout into the MySQL server.

package com.memeanalytics.storm_kafka_mysql;

import java.sql.PreparedStatement;

import backtype.storm.tuple.Tuple;

public class MySqlDump {
 private MySqlConnection conn;
 
 public MySqlDump(String ip, String database, String username, String password)
 {
  conn = new MySqlConnection(ip, database, username, password);
  conn.open();
 }
 
 public void persist(Tuple tuple)
 {
  PreparedStatement statement=null;
  try{
   statement = conn.getConnection().prepareStatement("insert into mylogtable (id, dataitem) values (default, ?)");
   statement.setString(1, tuple.getString(0));
   
   statement.executeUpdate();
  }catch(Exception ex)
  {
   ex.printStackTrace();
  }finally
  {
   if(statement != null)
   {
    try{
     statement.close();
    }catch(Exception ex)
    {
     ex.printStackTrace();
    }
   }
  }
 }
 
 public void close()
 {
  conn.close();
 }
}

The above classes opens the mysql connection in its constructor, and will saves the tuple via its persist() method, it also has a close() method which can be invoked to close the mysql connection.

Before we proceed further, we need to create the necessary database and datatable in mysql so that data can be inserted into. For demo purpose, we create a very basic datatable named "mylogtable" in a database named "mylog". To do this, access the mysql server by running the command:

> mysql -u [username] -p

Once logged into the mysql, run the following mysql queries to create the mylogtable:

> create database mylog;
> user mylog;
> create table mylogtable( id INT NOT NULL AUTO_INCREMENT, dataitem VARCHAR(255) NOT NULL, PRIMARY KEY(id));

Storm Bolt to persists data

Once this is done. We are ready to create a Storm bolt which route the received tuple from KafkaSpout to the MySqlDump which saves it to the database:

package com.memeanalytics.storm_kafka_mysql;

import java.util.Map;

import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Tuple;

public class MySqlDumpBolt extends BaseBasicBolt{

 private static final long serialVersionUID = 1L;
 private MySqlDump mySqlDump;

 @Override
 public void prepare(Map stormConf, TopologyContext context)
 {
  mySqlDump=new MySqlDump("localhost", "mylog","root","[username]");
 }
 
 public void execute(Tuple input, BasicOutputCollector collector) {
  // TODO Auto-generated method stub
  mySqlDump.persist(input);
  //System.out.println(input);
 }

 public void declareOutputFields(OutputFieldsDeclarer declarer) {
  // TODO Auto-generated method stub
  
 }
 
 @Override
    public void cleanup() {
  mySqlDump.close();
    }  

}

As can be seen above, the bolt opens the mysql database connection in its prepare() method and close the mysql database connection in its cleanup() method. it also persists the tuple received into mysql datatable in its execute() method.

Submit Topology in main()

Now we can implement the main() method to submit a topology. For simplicity, we only uses local cluster:

package com.memeanalytics.storm_kafka_mysql;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.spout.SchemeAsMultiScheme;
import backtype.storm.topology.TopologyBuilder;
import storm.kafka.KafkaSpout;
import storm.kafka.SpoutConfig;
import storm.kafka.StringScheme;
import storm.kafka.ZkHosts;

public class App 
{
    public static void main( String[] args ) throws AlreadyAliveException, InvalidTopologyException
    {
        ZkHosts zkHosts=new ZkHosts("192.168.2.4:2181");
        
        String topic="test-topic";
        String consumer_group_id="id7";
        
        SpoutConfig kafkaConfig=new SpoutConfig(zkHosts, topic, "", consumer_group_id);
        
        kafkaConfig.forceFromStart=true;
        kafkaConfig.scheme=new SchemeAsMultiScheme(new StringScheme());
        
        KafkaSpout kafkaSpout=new KafkaSpout(kafkaConfig);
        
        TopologyBuilder builder=new TopologyBuilder();
        builder.setSpout("KafkaSpout", kafkaSpout);
        builder.setBolt("MySqlBolt", new MySqlDumpBolt()).globalGrouping("KafkaSpout");
        
        LocalCluster cluster=new LocalCluster();
        
        Config config=new Config();
        
        cluster.submitTopology("MySqlDemoTopology", config, builder.createTopology());
        
        try{
         Thread.sleep(10000);
        }catch(InterruptedException ex)
        {
         ex.printStackTrace();
        }
        
        cluster.killTopology("MySqlDemoTopology");
        cluster.shutdown();
    }
}

Execute Maven project

Now navigate to the project root folder and run the following command:

> mvn exec:java -Dmain.class=com.memeanalytics.storm_kafka_mysql.App

Note that if you change the ${main.class} to com.memeanalytics.storm_kafka_mysql.App, that you don't need to add the -Dmain.class arguments in the above command.