Code cleanup

storm-ml/.classpath

@@ -1,7 +1,7 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <classpath>
 	<classpathentry kind="src" output="target/classes" path="src/main/java"/>
-	<classpathentry excluding="**" kind="src" output="target/classes" path="src/main/clojure"/>
+	<classpathentry excluding="**" kind="src" output="target/classes" path="src/main/resources"/>
 	<classpathentry kind="src" output="target/test-classes" path="src/test/java"/>
 	<classpathentry excluding="**" kind="src" output="target/test-classes" path="src/test/resources"/>
 	<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER/org.eclipse.jdt.internal.debug.ui.launcher.StandardVMType/J2SE-1.5"/>

storm-ml/src/main/java/com/twitter/MainOnlineTopology.java

@@ -1,40 +1,16 @@
 package com.twitter;
 
-import java.io.File;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Scanner;
-
 import backtype.storm.Config;
 import backtype.storm.LocalCluster;
 import backtype.storm.topology.TopologyBuilder;
-import backtype.storm.tuple.Values;
 import backtype.storm.utils.Utils;
 
 import com.twitter.storm.primitives.LocalLearner;
 import com.twitter.storm.primitives.TrainingSpout;
-import com.twitter.util.MathUtil;
 
 public class MainOnlineTopology {
 
-    public static List<List<Object>> readExamples(String fileName) throws IOException {
-        Scanner in = new Scanner(new File(fileName));
-        List<List<Object>> tupleList = new ArrayList<List<Object>>();
-        while (in.hasNext()) {
-            String line = in.nextLine();
-            tupleList.add(new Values(line));
-        }
-        in.close();
-        return tupleList;
-    }
-
     public static void main(String[] args) throws Exception {
-        int dimension = MathUtil.nextLikelyPrime(10);
-        System.out.println("Using dimension: " + dimension);
-
-        // Map exampleMap = new HashMap<Integer, List<List<Object>>>();
-        // exampleMap.put(0, readExamples(args[0]));
 
         TopologyBuilder builder = new TopologyBuilder();
         builder.setSpout("example_spitter", new TrainingSpout());
@@ -47,6 +23,5 @@ public static void main(String[] args) throws Exception {
         cluster.killTopology("test");
         cluster.shutdown();
 
-        // builder.setBolt("local_learner", new LocalLearner(dimension), 1).customGrouping(spout, grouping);
     }
 }

storm-ml/src/main/java/com/twitter/algorithms/Learner.java

@@ -18,7 +18,7 @@ public class Learner implements Serializable {
     int numMisclassified = 0;
     double totalLoss = 0.0;
     double gradientSum = 0.0;
-    protected double learningRate = 1.0;
+    protected double learningRate = 0.0;
 
     public Learner(int dimension) {
         weights = new double[dimension];
@@ -69,6 +69,10 @@ protected void updateStats(Example example, int prediction) {
         totalLoss += lossFunction.get(example, prediction);
     }
 
+    public void setWeights(double[] weights) {
+        this.weights = weights;
+    }
+
     public void displayStats() {
         if (numExamples == 0) {
             System.out.println("No examples seen so far.");

storm-ml/src/main/java/com/twitter/data/Example.java

@@ -17,47 +17,6 @@ public Example(int dimension) {
         isLabeled = false;
     }
 
-    /**
-     * 
-     * @param example
-     *            string representation of an example [+1,-1] | tag | importance | extra_info | feature:value pairs
-     */
-    public void parseFrom(String example, HashFunction hashFunction) {
-        int dimension = x.length;
-        example = example.trim();
-        String[] toks = example.split("\\|");
-        for (int i = 0; i < toks.length; i++) {
-            toks[i] = toks[i].trim();
-        }
-        try {
-            if (toks[0].equals("-1") || toks[0].equals("+1") || toks[0].equals("1") || toks[0].equals("0")) {
-                // label = Integer.parseInt(toks[0]);
-                isLabeled = true;
-            }
-            tag = toks[1];
-            importance = 1.0;
-            if (!toks[2].isEmpty()) {
-                importance = Double.parseDouble(toks[2]);
-            }
-            String extraInfo = toks[3];
-            // TODO (Delip): parse extraInfo
-            for (String fv : toks[4].split("\\s+")) {
-                String[] tmp = fv.split(":");
-                String feature = tmp[0];
-                double value = 1.0;
-                if (tmp.length == 2) {
-                    value = Double.parseDouble(tmp[1]);
-                }
-                int index = hashFunction.hash(feature, 0) % dimension;
-                x[index] += value;
-            }
-        } catch (Throwable e) {
-            System.err.println("Error Parsing:\n" + example);
-            e.printStackTrace();
-            return;
-        }
-    }
-
     public String toString() {
         return label + ":" + Arrays.toString(x);
     }

storm-ml/src/main/java/com/twitter/storm/primitives/LocalLearner.java

@@ -14,7 +14,6 @@
 import backtype.storm.transactional.ICommitter;
 import backtype.storm.tuple.Fields;
 import backtype.storm.tuple.Tuple;
-import backtype.storm.tuple.Values;
 
 import com.twitter.algorithms.Learner;
 import com.twitter.data.Example;
@@ -41,6 +40,10 @@ public LocalLearner(int dimension, Learner onlinePerceptron) {// , HashAll hashA
         this.learner = onlinePerceptron;
         // this.hashFunction = hashAll;
         weightVector = new double[dimension];
+        weightVector = new double[dimension];
+        weightVector[0] = -6.8;
+        weightVector[1] = -0.8;
+        learner.setWeights(weightVector);
     }
 
     public void execute(Tuple tuple) {
@@ -57,17 +60,6 @@ public void execute(Tuple tuple) {
         // buffer.add(example);
     }
 
-    public void finishBatch() {
-        if (buffer.size() == 0)
-            return;
-        learner.initWeights(weightVector);
-        for (Example e : buffer) {
-            learner.update(e, 1);
-        }
-
-        collector.emit(new Values(id, learner.getWeights(), learner.getParallelUpdateWeight()));
-    }
-
     public void declareOutputFields(OutputFieldsDeclarer declarer) {
         declarer.declare(new Fields("id", "weight_vector", "parallel_update_weights"));
     }