updates for new post on machine learning with kafka streams.

Author: bbejeck

README.md

@@ -1,7 +1,10 @@
 #kafka-streams
 This is the repository for the examples of using Kafka streams covered in the blog posts: 
 
-*    [Introducing The Kafka Processor API](codingjunkie.net/kafka-processor-part1/)
+ *   [Kafka Streams - The Processor API](http://codingjunkie.net/kafka-processor-part1/)
+ *   [Kafka Streams - The KStreams API](http://codingjunkie.net/kafka-streams-part2/)
+ *   [Machine Learning with Kafka Streams](http://codingjunkie.net/kafka-streams-machine-learning/)
+
 
 ## Requirements to build this project
 
@@ -10,20 +13,20 @@ This is the repository for the examples of using Kafka streams covered in the bl
 
 ## Requirements to run the examples
 
-1.    [kafka](https://github.com/apache/kafka) version kafka_2.11-0.10.0.0-SNAPSHOT see the section marked "Running a task on a particular version of Scala"
+1.    [kafka](https://github.com/apache/kafka) version kafka_2.11-0.10.1.0-SNAPSHOT see the section marked "Running a task on a particular version of Scala"
 2.    The [json-data-generator](https://github.com/acesinc/json-data-generator) from [ACES,Inc](http://acesinc.net/) 
 
 
 ## Dependencies included in repository
 As the Kafka Streams/Processor API is a work in progress, this repo includes the following jar files known to work with the examples
 
-1.    kafka-streams-0.10.0.0-SNAPSHOT.jar
-2.    kafka-clients-0.10.0.0-SNAPSHOT.jar
+1.    kafka-streams-0.10.1.0-SNAPSHOT.jar
+2.    kafka-clients-0.10.1.0-SNAPSHOT.jar
 
 ## Setup Instructions
 
-#### Extact the kafka_2.11-0.10.0.0-SNAPSHOT.tgz file ####
-    tar -xvzf kafka_2.11-0.10.0.0-SNAPSHOT.tgz
+#### Extact the kafka_2.11-0.10.1.0-SNAPSHOT.tgz file ####
+    tar -xvzf kafka_2.11-0.10.1.0-SNAPSHOT.tgz
 
 #### Install the Json-Data-Generator  
 Download the latest [json-data-generator release](https://github.com/acesinc/json-data-generator/releases) and follow the install instructions [here](http://acesinc.net/introducing-a-streaming-json-data-generator/)
@@ -64,6 +67,13 @@ Start zookeeper and kafka
      java -jar json-data-generator-1.2.0 stock-transactions-config.json
      cd kafka-streams
      ./gradlew runStockProcessor | runStockStreams
+     
+### Running the Twitter KStreams Language Classification Example ###
+    rename src/main/resources/twitter-app.properties.template to twitter-app.properties 
+    fill out the properties file with all the required values
+    
+    cd kafka-streams
+    ./gradlew runTwitterKstreamNLP 
 
 ### Viewing the results of the purchase streaming examples ###
     cd kafka_install-dir/bin
@@ -72,4 +82,8 @@ Start zookeeper and kafka
 ### Viewing the results of the stock-trading streaming examples ###
     cd kafka_install-dir/bin
     ./kafka-console-consumer --topic [stocks-out|transaction-summary] --zookeeper localhost:2181
+    
+### Viewing the results of the Twitter KStreams Language Classification Example ###
+    cd kafka_install-dir/bin
+    ./kafka-console-consumer --topic [english|french|spanish] --zookeeper localhost:2181    
           

bin/create-topics.sh

@@ -5,7 +5,7 @@ KAFKA_HOME=$1
 ZK_HOST=$2
 ZK_PORT=$3
 
-topics="src-topic patterns rewards purchases stocks stocks-out transaction-summary"
+topics="src-topic patterns rewards purchases stocks stocks-out transaction-summary twitterData english french spanish"
 
 for topic in ${topics}; do
      echo "attempting to create topic ${topic}"

build.gradle

@@ -32,6 +32,11 @@ task runStockStreams(type: JavaExec){
     main = "bbejeck.streams.stocks.StocksKafkaStreamsDriver"
 }
 
+task runTwitterKstreamNLP(type: JavaExec) {
+    classpath sourceSets.main.runtimeClasspath
+    main = "bbejeck.streams.twitter.TwitterKStreamNLPDriver"
+}
+
 
 
 
@@ -51,6 +56,9 @@ dependencies {
     compile group: 'org.apache.commons', name: 'commons-lang3', version:'3.1'
     compile group: 'com.101tec', name:'zkclient', version:'0.7'
     compile group: 'com.yammer.metrics', name:'metrics-core', version:'2.2.0'
-    compile files('libs/kafka-streams-0.10.0.0-SNAPSHOT.jar','libs/kafka-clients-0.10.0.0-SNAPSHOT.jar','libs/rocksdbjni-4.1.0.jar')
+    compile group: 'com.twitter', name:'hbc-core', version:'2.2.0'
+    compile group: 'de.julielab', name: 'aliasi-lingpipe', version:'4.1.0'
+    compile group: 'org.apache.commons', name:'commons-lang3', version:'3.4'
+    compile files('libs/kafka-streams-0.10.1.0-SNAPSHOT.jar','libs/kafka-clients-0.10.1.0-SNAPSHOT.jar','libs/rocksdbjni-4.1.0.jar')
     testCompile group: 'junit', name: 'junit', version:'4.11'
 }

libs/kafka-clients-0.10.0.0-SNAPSHOT.jar libs/kafka-clients-0.10.1.0-SNAPSHOT.jar

@@ -72,4 +72,14 @@ public void setTimeStamp(Date timeStamp) {
         this.timeStamp = timeStamp;
     }
 
+    @Override
+    public String toString() {
+        return "StockTransaction{" +
+                "symbol='" + symbol + '\'' +
+                ", type='" + type + '\'' +
+                ", shares=" + shares +
+                ", amount=" + amount +
+                ", timeStamp=" + timeStamp +
+                '}';
+    }
 }

libs/kafka-streams-0.10.0.0-SNAPSHOT.jar

@@ -42,4 +42,13 @@ public StockTransactionCollector add(StockTransaction transaction){
         return this;
     }
 
+    @Override
+    public String toString() {
+        return "StockTransactionCollector{" +
+                "amount=" + amount +
+                ", tickerSymbol='" + tickerSymbol + '\'' +
+                ", sharesPurchased=" + sharesPurchased +
+                ", sharesSold=" + sharesSold +
+                '}';
+    }
 }

libs/kafka-streams-0.10.1.0-SNAPSHOT.jar

@@ -0,0 +1,63 @@
+/*
+ * Copyright 2016 Bill Bejeck
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package bbejeck.model;
+
+/**
+ * User: Bill Bejeck
+ * Date: 4/21/16
+ * Time: 8:38 PM
+ */
+public class Tweet {
+
+    private String id;
+    private String text;
+    private String language;
+
+    public String getId() {
+        return id;
+    }
+
+    public void setId(String id) {
+        this.id = id;
+    }
+
+    public String getText() {
+        return text;
+    }
+
+    public void setText(String text) {
+        this.text = text;
+    }
+
+    public String getLanguage() {
+        return language;
+    }
+
+    public void setLanguage(String language) {
+        this.language = language;
+    }
+
+
+    @Override
+    public String toString() {
+        return "Tweet{" +
+                "id='" + id + '\'' +
+                ", text='" + text + '\'' +
+                ", language='" + language + '\'' +
+                '}';
+    }
+}

src/main/java/bbejeck/model/StockTransaction.java

@@ -0,0 +1,96 @@
+/*
+ * Copyright 2016 Bill Bejeck
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package bbejeck.nlp;
+
+import com.aliasi.classify.Classification;
+import com.aliasi.classify.Classified;
+import com.aliasi.classify.DynamicLMClassifier;
+import com.aliasi.lm.NGramBoundaryLM;
+
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * User: Bill Bejeck
+ * Date: 4/20/16
+ * Time: 10:01 PM
+ */
+public class Classifier {
+
+    private DynamicLMClassifier<NGramBoundaryLM> classifier;
+    private int maxCharNGram = 3;
+    private String trainingDataDelimiter;
+
+    public Classifier(String trainingDataDelimiter) {
+        this.trainingDataDelimiter = trainingDataDelimiter;
+    }
+
+    public Classifier(){
+         this("#");
+    }
+
+    public void train(File trainingData)  {
+        Set<String> categorySet = new HashSet<>();
+        List<String[]> annotatedData =  new ArrayList<>();
+        fillCategoriesAndAnnotatedData(trainingData, categorySet, annotatedData);
+        trainClassifier(categorySet, annotatedData);
+    }
+
+    private void trainClassifier(Set<String> categorySet, List<String[]> annotatedData){
+        String[] categories = categorySet.toArray(new String[0]);
+         classifier = DynamicLMClassifier.createNGramBoundary(categories,maxCharNGram);
+        for (String[] row: annotatedData) {
+            String actualClassification = row[0];
+            String text = row[1];
+            Classification classification = new Classification(actualClassification);
+            Classified<CharSequence> classified = new Classified<>(text,classification);
+            classifier.handle(classified);
+        }
+    }
+
+
+    private void fillCategoriesAndAnnotatedData(File trainingData,
+                                                Set<String> categorySet,
+                                                List<String[]> annotatedData) {
+        try (BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream(trainingData)))) {
+
+            String line = reader.readLine();
+            while (line != null) {
+                String[] data = line.split(trainingDataDelimiter);
+                categorySet.add(data[0]);
+                annotatedData.add(data);
+                line = reader.readLine();
+            }
+
+        } catch (IOException e){
+            throw new RuntimeException(e);
+        }
+    }
+
+
+    public String classify(String text){
+        return  classifier.classify(text.trim()).bestCategory().toLowerCase();
+    }
+
+}

src/main/java/bbejeck/model/StockTransactionCollector.java

@@ -0,0 +1,79 @@
+/*
+ * Copyright 2016 Bill Bejeck
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package bbejeck.nlp;
+
+import com.aliasi.classify.Classification;
+import com.aliasi.classify.Classified;
+import com.aliasi.classify.DynamicLMClassifier;
+import com.aliasi.lm.NGramBoundaryLM;
+
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.InputStreamReader;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * User: Bill Bejeck
+ * Date: 4/9/16
+ * Time: 6:24 PM
+ */
+public class LingPipeTester {
+
+    public static void main(String[] args) throws  Exception {
+        File trainingData = new File("src/main/resources/kafkaStreamsTwitterTrainingData_clean.csv");
+        int maxCharNGram = 3;
+
+        BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream(trainingData)));
+        Set<String> categorySet = new HashSet<>();
+        List<String[]> annotatedData =  new ArrayList<>();
+        String line = reader.readLine();
+        while (line !=null){
+              String[] data = line.split("#");
+              categorySet.add(data[0]);
+              annotatedData.add(data);
+              line = reader.readLine();
+        }
+        System.out.println("read in all data");
+        reader.close();
+        String[] categories = categorySet.toArray(new String[0]);
+
+        DynamicLMClassifier<NGramBoundaryLM> classifier
+                = DynamicLMClassifier.createNGramBoundary(categories,maxCharNGram);
+        for (String[] row: annotatedData) {
+            String truth = row[0];
+            String text = row[1];
+            Classification classification = new Classification(truth);
+            Classified<CharSequence> classified = new Classified<>(text,classification);
+            classifier.handle(classified);
+        }
+        System.out.println("training complete");
+
+        reader = new BufferedReader(new InputStreamReader(System.in));
+
+        System.out.println("enter text, I'll tell you the language");
+        String text;
+        while (!(text = reader.readLine()).equalsIgnoreCase("quit")) {
+            Classification classification = classifier.classify(text);
+            System.out.println("Entered -> " + text);
+            System.out.println("lang -> " + classification.bestCategory());
+        }
+    }
+}

src/main/java/bbejeck/model/Tweet.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2016 Bill Bejeck
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package bbejeck.processor.twitter;
+
+import bbejeck.model.Tweet;
+import org.apache.kafka.streams.processor.AbstractProcessor;
+
+/**
+ * User: Bill Bejeck
+ * Date: 4/21/16
+ * Time: 9:10 PM
+ */
+public class TwitterClassificationProcessor extends AbstractProcessor<String, Tweet> {
+
+
+
+    @Override
+    public void process(String s, Tweet tweet) {
+
+    }
+}