Showing posts with label sentiment. Show all posts
Showing posts with label sentiment. Show all posts
Wednesday, September 20, 2017
Trident ML Indexing Sentiment Classification results with ElasticSearch
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-MLs 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;The tuples emitted by the spout contains two fields: "text" and "label", the label is ignored, we are going to have the Trident-MLs TweetSentimentClassifier predict the sentiment label for us instead.
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");
}
}
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;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.
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();
}
}
}
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;Once it is completed, run the following command in the project root folder:
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();
}
}
> mvn compile exec:java
download file now
Friday, September 15, 2017
Twitter Sentiment Analysis of March Madness
Twitter Sentiment Analysis of March Madness
With March Madness officially over (and congrats to the Louisville Cardinals on a great season; thanks for winning my bracket pool), I thought I should pass along this article I came across regarding an application of Twitter sentiment analysis in the world of sports. The article at this link: http://www.vertica.com/2013/03/20/a-method-to-the-march-madness/ details how researchers used Twitter data to do sentiment analysis about NCAA basketball. Their model and results were presented recently at the MIT Sloan Sports Conference (which I really want to attend some day) in order to see if Twitter sentiment could predict the level of success of certain teams. The researchers hypothesized that those teams or players with a large number of tweets about them were more likely to be more successful due to the large number of people talking about them. Unfortanately, since the Sloan Conference was held before the actual start of what is commonly know as "March Madness" aka the NCAA Mens Basketball Tournament, the Twitter data used was an approximately one-week sample spanning the end of the regular season and beginning of the conference tournament games.
Michigans Trey Burke, Source: http://isportsweb.com/2011/11/15/michigan-basketball-trey-burke-leads-wolverines-victory/
The researchers limited the tweets they looked at to teams which were ranked in the top 25 of the AP poll at the time the data was collected, as well as top scorers from across college basketball. The researchers were able to data-mine almost 500,000 tweets in the week-long period and they show some interesting results when breaking the sentiment analysis down by team and player. Unsurprisingly, Michigans Trey Burke was the leader in positive sentiment analysis during the time period. This should not be a ground-breaking revelation because at the conclusion of the season he was named the National Player of the Year by the AP, and anyone who watched him carry Michigan to the national title game the past 2+ weeks knows how good he is. Apparently though, being good means getting a lot of Twitter love. Most of the players heading the largest sentiment graphic were well-known stars, although I did notice an absence of players from smaller schools having big years, such as Creightons Doug McDermott. I guess when you are not on TV every week the Twitterverse isnt all that interested in you. The other player I found a surprising amount of love for was Kentucky center Nerlens Noel. For those of you who dont know, Noel injured a knee earlier in the year and was not even playing at the time the data was mined, but still got a surprising amount of sentiment. My only guess is that UK fans were collectively whining about how the would have made the tourney had he not been injured.
Sad UK Fan, Source: http://www.crimsoncast.com/2012/03/pre-game-meal-kentucky-wants-revenge/imagescau0mx6c-2/
Data about individual teams is only briefly mentioned in the article, with only a supporting graphic showing the sentiment for the Kansas Jayhawks, but the authors did include a note that traditional powerhouse teams led the way in tweet volume, which was not at all surprising. The final thing that really surprised me was the volume of tweets coming from overseas, particularly London. The United Kingdom is not generally know as a basketball fan country, sticking mostly to soccer, but tweets out of London outpaced many major American cities, which was surprising to say the least. Even more impressive is that most people there are asleep when games are being played over here, meaning the Brits are dedicated enough to check up on happenings around college hoops the next day, then tweet about it retroactively. I got some cool insights out of this article, and it concludes with a challenge to all of us to try and use their HP Vertica platform to generate a model combining Twitter sentiment analysis and statistics to try and predict the winner. Its a little late to try it out until next year, but I will be keeping a close eye on their blog to see if anyone gave it their best shot.
download file now
Sunday, August 6, 2017
Trident ML Sentiment Analysis Classifier
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 dont 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
download file now
Subscribe to:
Posts (Atom)