SAMOA-34: Fix Bagging
diff --git a/samoa-api/src/main/java/org/apache/samoa/learners/classifiers/ensemble/Bagging.java b/samoa-api/src/main/java/org/apache/samoa/learners/classifiers/ensemble/Bagging.java
index 7355b1a..43bc07c 100644
--- a/samoa-api/src/main/java/org/apache/samoa/learners/classifiers/ensemble/Bagging.java
+++ b/samoa-api/src/main/java/org/apache/samoa/learners/classifiers/ensemble/Bagging.java
@@ -24,8 +24,6 @@
* License
*/
-import com.google.common.collect.ImmutableSet;
-
import java.util.Set;
import org.apache.samoa.core.Processor;
@@ -34,10 +32,13 @@
import org.apache.samoa.learners.classifiers.trees.VerticalHoeffdingTree;
import org.apache.samoa.topology.Stream;
import org.apache.samoa.topology.TopologyBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import com.github.javacliparser.ClassOption;
import com.github.javacliparser.Configurable;
import com.github.javacliparser.IntOption;
+import com.google.common.collect.ImmutableSet;
/**
* The Bagging Classifier by Oza and Russell.
@@ -46,6 +47,7 @@
/** The Constant serialVersionUID. */
private static final long serialVersionUID = -2971850264864952099L;
+ private static final Logger logger = LoggerFactory.getLogger(Bagging.class);
/** The base learner option. */
public ClassOption baseLearnerOption = new ClassOption("baseLearner", 'l',
@@ -58,11 +60,8 @@
/** The distributor processor. */
private BaggingDistributorProcessor distributorP;
- /** The training stream. */
- private Stream testingStream;
-
- /** The prediction stream. */
- private Stream predictionStream;
+ /** The input streams for the ensemble, one per member. */
+ private Stream[] ensembleStreams;
/** The result stream. */
protected Stream resultStream;
@@ -70,45 +69,57 @@
/** The dataset. */
private Instances dataset;
- protected Learner classifier;
+ protected Learner[] ensemble;
protected int parallelism;
/**
* Sets the layout.
+ *
+ * @throws Exception
*/
protected void setLayout() {
-
- int sizeEnsemble = this.ensembleSizeOption.getValue();
+ int ensembleSize = this.ensembleSizeOption.getValue();
distributorP = new BaggingDistributorProcessor();
- distributorP.setSizeEnsemble(sizeEnsemble);
- this.builder.addProcessor(distributorP, 1);
+ distributorP.setEnsembleSize(ensembleSize);
+ builder.addProcessor(distributorP, 1);
// instantiate classifier
- classifier = (Learner) this.baseLearnerOption.getValue();
- classifier.init(builder, this.dataset, sizeEnsemble);
+ ensemble = new Learner[ensembleSize];
+ for (int i = 0; i < ensembleSize; i++) {
+ try {
+ ensemble[i] = (Learner) ClassOption.createObject(baseLearnerOption.getValueAsCLIString(),
+ baseLearnerOption.getRequiredType());
+ } catch (Exception e) {
+ logger.error("Unable to create members of the ensemble. Please check your CLI parameters");
+ e.printStackTrace();
+ throw new IllegalArgumentException(e);
+ }
+ ensemble[i].init(builder, this.dataset, 1); // sequential
+ }
PredictionCombinerProcessor predictionCombinerP = new PredictionCombinerProcessor();
- predictionCombinerP.setSizeEnsemble(sizeEnsemble);
+ predictionCombinerP.setEnsembleSize(ensembleSize);
this.builder.addProcessor(predictionCombinerP, 1);
// Streams
- resultStream = this.builder.createStream(predictionCombinerP);
+ resultStream = builder.createStream(predictionCombinerP);
predictionCombinerP.setOutputStream(resultStream);
- for (Stream subResultStream : classifier.getResultStreams()) {
- this.builder.connectInputKeyStream(subResultStream, predictionCombinerP);
+ for (Learner member : ensemble) {
+ for (Stream subResultStream : member.getResultStreams()) { // a learner can have multiple output streams
+ this.builder.connectInputKeyStream(subResultStream, predictionCombinerP); // the key is the instance id to combine predictions
+ }
}
- testingStream = this.builder.createStream(distributorP);
- this.builder.connectInputKeyStream(testingStream, classifier.getInputProcessor());
+ ensembleStreams = new Stream[ensembleSize];
+ for (int i = 0; i < ensembleSize; i++) {
+ ensembleStreams[i] = builder.createStream(distributorP);
+ builder.connectInputShuffleStream(ensembleStreams[i], ensemble[i].getInputProcessor()); // connect streams one-to-one with ensemble members (the type of connection does not matter)
+ }
- predictionStream = this.builder.createStream(distributorP);
- this.builder.connectInputKeyStream(predictionStream, classifier.getInputProcessor());
-
- distributorP.setOutputStream(testingStream);
- distributorP.setPredictionStream(predictionStream);
+ distributorP.setOutputStreams(ensembleStreams);
}
/** The builder. */
diff --git a/samoa-api/src/main/java/org/apache/samoa/learners/classifiers/ensemble/BaggingDistributorProcessor.java b/samoa-api/src/main/java/org/apache/samoa/learners/classifiers/ensemble/BaggingDistributorProcessor.java
index 33615db..6c88d94 100644
--- a/samoa-api/src/main/java/org/apache/samoa/learners/classifiers/ensemble/BaggingDistributorProcessor.java
+++ b/samoa-api/src/main/java/org/apache/samoa/learners/classifiers/ensemble/BaggingDistributorProcessor.java
@@ -24,6 +24,7 @@
* License
*/
+import java.util.Arrays;
import java.util.Random;
import org.apache.samoa.core.ContentEvent;
@@ -38,19 +39,16 @@
*/
public class BaggingDistributorProcessor implements Processor {
- /**
- *
- */
private static final long serialVersionUID = -1550901409625192730L;
- /** The size ensemble. */
- private int sizeEnsemble;
+ /** The ensemble size. */
+ private int ensembleSize;
- /** The training stream. */
- private Stream trainingStream;
+ /** The stream ensemble. */
+ private Stream[] ensembleStreams;
- /** The prediction stream. */
- private Stream predictionStream;
+ /** Ramdom number generator. */
+ protected Random random = new Random(); //TODO make random seed configurable
/**
* On event.
@@ -60,38 +58,34 @@
* @return true, if successful
*/
public boolean process(ContentEvent event) {
- InstanceContentEvent inEvent = (InstanceContentEvent) event; // ((s4Event)event).getContentEvent();
- // InstanceEvent inEvent = (InstanceEvent) event;
+ InstanceContentEvent inEvent = (InstanceContentEvent) event;
if (inEvent.getInstanceIndex() < 0) {
- // End learning
- predictionStream.put(event);
+ // end learning
+ for (Stream stream : ensembleStreams)
+ stream.put(event);
return false;
}
if (inEvent.isTesting()) {
- Instance trainInst = inEvent.getInstance();
- for (int i = 0; i < sizeEnsemble; i++) {
- Instance weightedInst = trainInst.copy();
- // weightedInst.setWeight(trainInst.weight() * k);
- InstanceContentEvent instanceContentEvent = new InstanceContentEvent(
- inEvent.getInstanceIndex(), weightedInst, false, true);
- instanceContentEvent.setClassifierIndex(i);
- instanceContentEvent.setEvaluationIndex(inEvent.getEvaluationIndex());
- predictionStream.put(instanceContentEvent);
+ Instance testInstance = inEvent.getInstance();
+ for (int i = 0; i < ensembleSize; i++) {
+ Instance instanceCopy = testInstance.copy();
+ InstanceContentEvent instanceContentEvent = new InstanceContentEvent(inEvent.getInstanceIndex(), instanceCopy,
+ false, true);
+ instanceContentEvent.setClassifierIndex(i); //TODO probably not needed anymore
+ instanceContentEvent.setEvaluationIndex(inEvent.getEvaluationIndex()); //TODO probably not needed anymore
+ ensembleStreams[i].put(instanceContentEvent);
}
}
- /* Estimate model parameters using the training data. */
+ // estimate model parameters using the training data
if (inEvent.isTraining()) {
train(inEvent);
}
- return false;
+ return true;
}
- /** The random. */
- protected Random random = new Random();
-
/**
* Train.
*
@@ -99,104 +93,51 @@
* the in event
*/
protected void train(InstanceContentEvent inEvent) {
- Instance trainInst = inEvent.getInstance();
- for (int i = 0; i < sizeEnsemble; i++) {
+ Instance trainInstance = inEvent.getInstance();
+ for (int i = 0; i < ensembleSize; i++) {
int k = MiscUtils.poisson(1.0, this.random);
if (k > 0) {
- Instance weightedInst = trainInst.copy();
- weightedInst.setWeight(trainInst.weight() * k);
- InstanceContentEvent instanceContentEvent = new InstanceContentEvent(
- inEvent.getInstanceIndex(), weightedInst, true, false);
+ Instance weightedInstance = trainInstance.copy();
+ weightedInstance.setWeight(trainInstance.weight() * k);
+ InstanceContentEvent instanceContentEvent = new InstanceContentEvent(inEvent.getInstanceIndex(),
+ weightedInstance, true, false);
instanceContentEvent.setClassifierIndex(i);
instanceContentEvent.setEvaluationIndex(inEvent.getEvaluationIndex());
- trainingStream.put(instanceContentEvent);
+ ensembleStreams[i].put(instanceContentEvent);
}
}
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.s4.core.ProcessingElement#onCreate()
- */
@Override
public void onCreate(int id) {
// do nothing
}
- /**
- * Gets the training stream.
- *
- * @return the training stream
- */
- public Stream getTrainingStream() {
- return trainingStream;
+ public Stream[] getOutputStreams() {
+ return ensembleStreams;
}
- /**
- * Sets the training stream.
- *
- * @param trainingStream
- * the new training stream
- */
- public void setOutputStream(Stream trainingStream) {
- this.trainingStream = trainingStream;
+ public void setOutputStreams(Stream[] ensembleStreams) {
+ this.ensembleStreams = ensembleStreams;
}
- /**
- * Gets the prediction stream.
- *
- * @return the prediction stream
- */
- public Stream getPredictionStream() {
- return predictionStream;
+ public int getEnsembleSize() {
+ return ensembleSize;
}
- /**
- * Sets the prediction stream.
- *
- * @param predictionStream
- * the new prediction stream
- */
- public void setPredictionStream(Stream predictionStream) {
- this.predictionStream = predictionStream;
+ public void setEnsembleSize(int ensembleSize) {
+ this.ensembleSize = ensembleSize;
}
- /**
- * Gets the size ensemble.
- *
- * @return the size ensemble
- */
- public int getSizeEnsemble() {
- return sizeEnsemble;
- }
-
- /**
- * Sets the size ensemble.
- *
- * @param sizeEnsemble
- * the new size ensemble
- */
- public void setSizeEnsemble(int sizeEnsemble) {
- this.sizeEnsemble = sizeEnsemble;
- }
-
- /*
- * (non-Javadoc)
- *
- * @see samoa.core.Processor#newProcessor(samoa.core.Processor)
- */
@Override
public Processor newProcessor(Processor sourceProcessor) {
BaggingDistributorProcessor newProcessor = new BaggingDistributorProcessor();
BaggingDistributorProcessor originProcessor = (BaggingDistributorProcessor) sourceProcessor;
- if (originProcessor.getPredictionStream() != null) {
- newProcessor.setPredictionStream(originProcessor.getPredictionStream());
+ if (originProcessor.getOutputStreams() != null) {
+ newProcessor.setOutputStreams(Arrays.copyOf(originProcessor.getOutputStreams(),
+ originProcessor.getOutputStreams().length));
}
- if (originProcessor.getTrainingStream() != null) {
- newProcessor.setOutputStream(originProcessor.getTrainingStream());
- }
- newProcessor.setSizeEnsemble(originProcessor.getSizeEnsemble());
+ newProcessor.setEnsembleSize(originProcessor.getEnsembleSize());
/*
* if (originProcessor.getLearningCurve() != null){
* newProcessor.setLearningCurve((LearningCurve)
@@ -204,5 +145,4 @@
*/
return newProcessor;
}
-
}
diff --git a/samoa-api/src/main/java/org/apache/samoa/learners/classifiers/ensemble/PredictionCombinerProcessor.java b/samoa-api/src/main/java/org/apache/samoa/learners/classifiers/ensemble/PredictionCombinerProcessor.java
index 2e5f335..76e84f8 100644
--- a/samoa-api/src/main/java/org/apache/samoa/learners/classifiers/ensemble/PredictionCombinerProcessor.java
+++ b/samoa-api/src/main/java/org/apache/samoa/learners/classifiers/ensemble/PredictionCombinerProcessor.java
@@ -33,14 +33,14 @@
import org.apache.samoa.topology.Stream;
/**
- * The Class PredictionCombinerProcessor.
+ * Combines predictions coming from an ensemble. Equivalent to a majority-vote classifier.
*/
public class PredictionCombinerProcessor implements Processor {
private static final long serialVersionUID = -1606045723451191132L;
/**
- * The size ensemble.
+ * The ensemble size.
*/
protected int ensembleSize;
@@ -73,7 +73,7 @@
*
* @return the ensembleSize
*/
- public int getSizeEnsemble() {
+ public int getEnsembleSize() {
return ensembleSize;
}
@@ -83,7 +83,7 @@
* @param ensembleSize
* the new size ensemble
*/
- public void setSizeEnsemble(int ensembleSize) {
+ public void setEnsembleSize(int ensembleSize) {
this.ensembleSize = ensembleSize;
}
@@ -143,7 +143,7 @@
if (originProcessor.getOutputStream() != null) {
newProcessor.setOutputStream(originProcessor.getOutputStream());
}
- newProcessor.setSizeEnsemble(originProcessor.getSizeEnsemble());
+ newProcessor.setEnsembleSize(originProcessor.getEnsembleSize());
return newProcessor;
}
diff --git a/samoa-api/src/main/java/org/apache/samoa/learners/classifiers/trees/VerticalHoeffdingTree.java b/samoa-api/src/main/java/org/apache/samoa/learners/classifiers/trees/VerticalHoeffdingTree.java
index ea7e53d..6534cee 100644
--- a/samoa-api/src/main/java/org/apache/samoa/learners/classifiers/trees/VerticalHoeffdingTree.java
+++ b/samoa-api/src/main/java/org/apache/samoa/learners/classifiers/trees/VerticalHoeffdingTree.java
@@ -20,8 +20,6 @@
* #L%
*/
-import com.google.common.collect.ImmutableSet;
-
import java.util.Set;
import org.apache.samoa.core.Processor;
@@ -41,6 +39,7 @@
import com.github.javacliparser.FlagOption;
import com.github.javacliparser.FloatOption;
import com.github.javacliparser.IntOption;
+import com.google.common.collect.ImmutableSet;
/**
* Vertical Hoeffding Tree.
@@ -172,7 +171,7 @@
public void setChangeDetector(ChangeDetector cd) {
this.changeDetector = cd;
}
-
+
static class LearningNodeIdGenerator {
// TODO: add code to warn user of when value reaches Long.MAX_VALUES
