User aggregators

examples/naive.boa

@@ -0,0 +1,172 @@
+p: Project = input;
+type fv = {a:int, b:int, c:int, d:int};
+type stats = {a_stat:float, b_stat:float, c_stat:float};
+type complete_stat = {avg: stats, dev: stats};
+type Data = {training: fv, testing: fv};
+splitRatio : float = 0.67;
+
+naive := function(vals : array of Data) : float {
+   train : array of fv;
+   test : array of fv;
+
+    spearated: map[int] of array of fv; # classified per value
+    summaries : map[int] of complete_stat;
+
+   # separate the training and testing datasets
+    foreach(i:int; def(vals[i])) {
+        if(def(train)) {
+           train = train + {vals[i].training};
+        } else {
+           train = {vals[i].training};
+        }
+        if(def(test)) {
+           test = test+ {vals[i].testing};
+        } else {
+           test =  {vals[i].testing};
+        }
+
+    }
+
+
+    # classify training datasets
+    foreach(i:int; def(train[i])) {
+        temp : array of fv = {train[i]};
+        if(!haskey(spearated, train[i].d)) {
+          spearated[train[i].d] = temp;
+        } else {
+          spearated[train[i].d] = spearated[train[i].d] + temp;
+        }
+    }
+
+    # all the classes
+    classes : array of int = keys(spearated);
+
+    # summarize data from training dataset
+    foreach(i:int; def(classes[i])) {
+          # calculate mean
+          feature_mean : stats = {0.0, 0.0, 0.0};
+          foreach(j:int; def(spearated[classes[i]][j])) {
+              feature_mean.a_stat = feature_mean.a_stat + spearated[classes[i]][j].a;
+              feature_mean.b_stat = feature_mean.b_stat + spearated[classes[i]][j].b;
+              feature_mean.c_stat = feature_mean.c_stat + spearated[classes[i]][j].c;
+          }
+          feature_mean.a_stat = feature_mean.a_stat / len(spearated[classes[i]]);
+          feature_mean.b_stat = feature_mean.b_stat / len(spearated[classes[i]]);
+          feature_mean.c_stat = feature_mean.c_stat / len(spearated[classes[i]]);
+
+
+          # calculate sd
+          feature_sd : stats = {0.0, 0.0, 0.0};
+          foreach(j:int; def(spearated[classes[i]][j])) {
+             feature_sd.a_stat = feature_sd.a_stat + (spearated[classes[i]][j].a - feature_mean.a_stat);
+             feature_sd.b_stat = feature_sd.b_stat + (spearated[classes[i]][j].b - feature_mean.b_stat);
+             feature_sd.c_stat = feature_sd.c_stat + (spearated[classes[i]][j].c - feature_mean.c_stat);
+          }
+          feature_sd.a_stat = sqrt(feature_sd.a_stat / len(spearated[classes[i]]));
+          feature_sd.b_stat = sqrt(feature_sd.b_stat / len(spearated[classes[i]]));
+          feature_sd.c_stat = sqrt(feature_sd.c_stat / len(spearated[classes[i]]));
+
+          # summarized a class
+          summaries[classes[i]] = {feature_mean, feature_sd};
+    }
+
+
+    predictions: array of int;
+    predictions = new(predictions, len(test), -1);
+
+    # predict for each test data
+    foreach(i:int; def(test[i])) {
+        probabilities : map[int] of float;
+        foreach(j: int; def(classes[j])) {
+          probabilities[classes[j]] = 1.0;
+          mean := summaries[classes[j]].avg;
+          deviation := summaries[classes[j]].dev;
+          probabilities[classes[j]] = probabilities[classes[j]] * (1/ (sqrt(2 * 3.14) * deviation.a_stat)) * (exp(-1 * ((pow((1.0 * test[i].a) - mean.a_stat, 2))/(2 * pow(deviation.a_stat, 2)))));
+          probabilities[classes[j]] = probabilities[classes[j]] * (1/ (sqrt(2 * 3.14) * deviation.a_stat)) * (exp(-1 * ((pow((1.0 * test[i].b) - mean.b_stat, 2))/(2 * pow(deviation.b_stat, 2)))));
+          probabilities[classes[j]] = probabilities[classes[j]] * (1/ (sqrt(2 * 3.14) * deviation.a_stat)) * (exp(-1 * ((pow((1.0 * test[i].c) - mean.c_stat, 2))/(2 * pow(deviation.c_stat, 2)))));
+       }
+
+        bestProb : float = 0;
+        bestLab : int = -1;
+        foreach(j: int; def(classes[j])) {
+          if ((bestLab == -1) || (bestProb < probabilities[classes[j]])) {
+            bestProb = probabilities[classes[j]];
+            bestLab = classes[j];
+          }
+        }
+        predictions[i] = bestLab;
+    }
+
+    correct : float = 0.0;
+    foreach(i:int; def(test[i])) {
+        if(predictions[i] == test[i].d) {
+          correct = correct + 1.0;
+        }
+    }
+    return correct/len(test) * 100;
+};
+
+scale := function(ast: int, method: int, class: int) : int {
+    total : int = 0;
+    if(ast > 1000) {
+       total++;
+    } if(method > 500) {
+        total++;
+    } if(class > 50) {
+      total++;
+    }
+    return total;
+};
+
+
+naive_bayes : output naive of Data;
+
+# count ast nodes
+
+astCount := 0;
+classCount := 0;
+methodCount := 0;
+visit(p, visitor {
+	# only look at the latest snapshot
+	before n: CodeRepository -> {
+		snapshot := getsnapshot(n);
+		foreach (i: int; def(snapshot[i]))
+			visit(snapshot[i]);
+		stop;
+	}
+	before node: Declaration -> {
+		if (node.kind == TypeKind.CLASS) {
+            classCount++;
+            foreach (i: int; node.methods[i]) {
+                methodCount++;
+            }
+		}
+	}
+	# by default, count all visited nodes
+	before _ -> astCount++;
+	# these nodes are not part of the AST, so do nothing when visiting
+	before Project, ChangedFile -> ;
+});
+
+
+
+dummy : fv = {0, 0, 0, 0};
+nondummy : fv = {astCount, methodCount, classCount, scale(astCount, methodCount, classCount)};
+data1: Data = {nondummy, dummy};
+data2: Data = {dummy, nondummy};
+if(rand() > splitRatio)
+    naive_bayes << data1;
+else
+    naive_bayes << data2;
+
+
+if(rand() > splitRatio)
+    naive_bayes << data1;
+else
+    naive_bayes << data2;
+
+
+if(rand() > splitRatio)
+    naive_bayes << data1;
+else
+    naive_bayes << data2;

src/java/boa/BoaEnumInterface.java

@@ -0,0 +1,5 @@
+package boa;
+
+public interface BoaEnumInterface {
+    Object getValue();
+}

src/java/boa/BoaTup.java

@@ -0,0 +1,12 @@
+package boa;
+
+import java.io.IOException;
+
+
+public interface BoaTup {
+    public String[] getValues();
+    public byte[] serialize(Object o) throws IOException;
+    public Object getValue(String f);
+    public String toString();
+    public String[] getFieldNames();
+}

src/java/boa/aggregators/Aggregator.java

@@ -18,10 +18,14 @@
 
 import java.io.IOException;
 
+import boa.BoaTup;
 import org.apache.hadoop.io.NullWritable;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.mapreduce.Reducer.Context;
 
+import java.util.HashSet;
+import java.util.Set;
+
 import boa.functions.BoaCasts;
 import boa.io.EmitKey;
 import boa.io.EmitValue;
@@ -38,6 +42,29 @@ public abstract class Aggregator {
 	private Context context;
 	private EmitKey key;
 	private boolean combining;
+	private final static Set<String> inBuiltAggs = new HashSet<String>();
+
+	static {
+		inBuiltAggs.add("sum");
+		inBuiltAggs.add("top");
+		inBuiltAggs.add("maximum");
+		inBuiltAggs.add("minimum");
+		inBuiltAggs.add("max");
+		inBuiltAggs.add("min");
+		inBuiltAggs.add("collection");
+		inBuiltAggs.add("mean");
+		inBuiltAggs.add("median");
+		inBuiltAggs.add("stdev");
+		inBuiltAggs.add("quantile");
+		inBuiltAggs.add("kurtosis");
+		inBuiltAggs.add("histogram");
+		inBuiltAggs.add("graphCSV");
+		inBuiltAggs.add("set");
+		inBuiltAggs.add("bottom");
+		inBuiltAggs.add("skewness");
+		inBuiltAggs.add("confidence");
+		inBuiltAggs.add("variance");
+	}
 
 	/**
 	 * Construct an Aggregator.
@@ -93,6 +120,13 @@ public void aggregate(final double data) throws IOException, InterruptedExceptio
 		this.aggregate(BoaCasts.doubleToString(data), null);
 	}
 
+	public void aggregate(final BoaTup data, final String metadata) throws IOException, InterruptedException, FinishedException {
+	}
+
+	public void aggregate(final BoaTup data) throws IOException, InterruptedException, FinishedException {
+		this.aggregate(data, null);
+	}
+
 	@SuppressWarnings("unchecked")
 	protected void collect(final String data, final String metadata) throws IOException, InterruptedException {
 		if (this.combining)
@@ -107,6 +141,10 @@ protected void collect(final String data) throws IOException, InterruptedExcepti
 		this.collect(data, null);
 	}
 
+	protected void collect(final BoaTup data) throws IOException, InterruptedException {
+		this.collect(data.toString(), null);
+	}
+
 	@SuppressWarnings("unchecked")
 	protected void collect(final long data, final String metadata) throws IOException, InterruptedException {
 		this.collect(BoaCasts.longToString(data), metadata);
@@ -157,4 +195,8 @@ public void setKey(final EmitKey key) {
 	public EmitKey getKey() {
 		return this.key;
 	}
+
+	public final static boolean isUserDefinedAggregator(String name) {
+		return !Aggregator.inBuiltAggs.contains(name);
+	}
 }

src/java/boa/aggregators/UserDefinedAggregator.java

@@ -0,0 +1,6 @@
+package boa.aggregators;
+
+
+@AggregatorSpec(name = "UserDefinedAgg", type = "UserDefined", canCombine = false)
+public abstract class UserDefinedAggregator extends Aggregator {
+}

src/java/boa/compiler/BoaCompiler.java

@@ -155,6 +155,8 @@ public void syntaxError(Recognizer<?, ?> recognizer, Object offendingSymbol, int
 
 					try {
 						if (!parserErrorListener.hasError) {
+							UserDefinedAggregators.setFileName(f.getName());
+							UserDefinedAggregators.setJobName("Job" + jobName);
 							new TypeCheckingVisitor().start(p, new SymbolTable());
 
 							final TaskClassifyingVisitor simpleVisitor = new TaskClassifyingVisitor();
@@ -241,6 +243,7 @@ public void syntaxError(Recognizer<?, ?> recognizer, Object offendingSymbol, int
 			st.add("jobs", jobs);
 			st.add("jobnames", jobnames);
 			st.add("combineTables", CodeGeneratingVisitor.combineAggregatorStrings);
+			st.add("userDeclAgg", CodeGeneratingVisitor.userAggregatorDeclStrings);
 			st.add("reduceTables", CodeGeneratingVisitor.reduceAggregatorStrings);
 			st.add("splitsize", isSimple ? 64 * 1024 * 1024 : 10 * 1024 * 1024);
 			if(DefaultProperties.localDataPath != null) {

src/java/boa/compiler/SymbolTable.java

@@ -22,6 +22,7 @@
 import java.util.*;
 import java.util.Map.Entry;
 
+import boa.aggregators.UserDefinedAggregator;
 import org.scannotation.AnnotationDB;
 
 import boa.aggregators.AggregatorSpec;
@@ -435,15 +436,28 @@ else if (aggregators.containsKey(name))
 	}
 
 	public List<Class<?>> getAggregators(final String name, final BoaType type) {
-		final List<Class<?>> aggregators = new ArrayList<Class<?>>();
-
-		if (type instanceof BoaTuple)
-			for (final BoaType subType : ((BoaTuple) type).getTypes())
-				aggregators.add(this.getAggregator(name, subType));
-		else
-			aggregators.add(this.getAggregator(name, type));
+		final List<Class<?>> searchResult = new ArrayList<Class<?>>();
+		if (possibleInBuiltAgg(name)) {
+			if (type instanceof BoaTuple)
+				searchResult.add(this.getAggregator(name, type));
+			else if (type instanceof BoaArray)
+				searchResult.add(this.getAggregator(name, ((BoaArray)type).getType()));
+			else
+				searchResult.add(this.getAggregator(name, type));
+		} else if (this.functions.hasFunction(name)) {
+			searchResult.add(UserDefinedAggregator.class);
+		}
+		return searchResult;
+	}
 
-		return aggregators;
+	private boolean possibleInBuiltAgg(String name) {
+		Set<String> names = aggregators.keySet();
+		for (final String entry: names) {
+			if(entry.contains(name)) {
+				return true;
+			}
+		}
+		return false;
 	}
 
 	private static void importFunction(final Method m) {