first checkin for heron ml

heron/mlmgr/src/main/java/org/apache/heron/LocalStormDoTask.java

@@ -0,0 +1,82 @@
+package org.apache.heron;
+
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.
+ */
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.samoa.topology.impl.StormSamoaUtils;
+import org.apache.samoa.topology.impl.StormTopology;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.apache.commons.configuration.Configuration;
+
+import backtype.storm.Config;
+import backtype.storm.utils.Utils;
+
+/**
+ * The main class to execute a SAMOA task in LOCAL mode in Storm.
+ * 
+ * @author Arinto Murdopo
+ * 
+ */
+public class LocalStormDoTask {
+
+  private static final Logger logger = LoggerFactory.getLogger(LocalStormDoTask.class);
+  private static final String EXECUTION_DURATION_KEY ="samoa.storm.local.mode.execution.duration";
+  private static final String SAMOA_STORM_PROPERTY_FILE_LOC ="samoa-storm.properties";
+  /**
+   * The main method.
+   * 
+   * @param args
+   *          the arguments
+   */
+  public static void main(String[] args) {
+
+    List<String> tmpArgs = new ArrayList<String>(Arrays.asList(args));
+
+    int numWorker = StormSamoaUtils.numWorkers(tmpArgs);
+
+    args = tmpArgs.toArray(new String[0]);
+
+    // convert the arguments into Storm topology
+    StormTopology stormTopo = StormSamoaUtils.argsToTopology(args);
+    String topologyName = stormTopo.getTopologyName();
+
+    Config conf = new Config();
+    // conf.putAll(Utils.readStormConfig());
+    conf.setDebug(false);
+
+    // local mode
+    conf.setMaxTaskParallelism(numWorker);
+
+    backtype.storm.LocalCluster cluster = new backtype.storm.LocalCluster();
+    cluster.submitTopology(topologyName, conf, stormTopo.getStormBuilder().createTopology());
+
+    // Read local mode execution duration from property file
+    Configuration stormConfig = StormSamoaUtils.getPropertyConfig(LocalStormDoTask.SAMOA_STORM_PROPERTY_FILE_LOC);
+    long executionDuration= stormConfig.getLong(LocalStormDoTask.EXECUTION_DURATION_KEY);
+    backtype.storm.utils.Utils.sleep(executionDuration * 1000);
+
+    cluster.killTopology(topologyName);
+    cluster.shutdown();
+
+  }
+}

heron/mlmgr/src/main/java/org/apache/heron/learners/AdaptiveLearner.java

@@ -0,0 +1,46 @@
+package org.apache.heron.learners;
+
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.
+ */
+
+import org.apache.samoa.moa.classifiers.core.driftdetection.ChangeDetector;
+
+/**
+ * The Interface Adaptive Learner. Initializing Classifier should initalize PI to connect the Classifier with the input
+ * stream and initialize result stream so that other PI can connect to the classification result of this classifier
+ */
+
+public interface AdaptiveLearner extends Learner{
+
+  /**
+   * Gets the change detector item.
+   * 
+   * @return the change detector item
+   */
+  public ChangeDetector getChangeDetector();
+
+  /**
+   * Sets the change detector item.
+   * 
+   * @param cd
+   *          the change detector item
+   */
+  public void setChangeDetector(ChangeDetector cd);
+
+}

heron/mlmgr/src/main/java/org/apache/heron/learners/ClassificationLearner.java

@@ -0,0 +1,26 @@
+package org.apache.heron.learners;
+
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.
+ */
+
+import org.apache.samoa.learners.Learner;
+
+public interface ClassificationLearner extends Learner {
+
+}

heron/mlmgr/src/main/java/org/apache/heron/learners/InstanceContent.java

@@ -0,0 +1,202 @@
+package org.apache.heron.learners;
+
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.
+ */
+
+/**
+ * License
+ */
+
+import net.jcip.annotations.Immutable;
+
+import org.apache.samoa.core.SerializableInstance;
+import org.apache.samoa.instances.Instance;
+
+import java.io.Serializable;
+
+/**
+ * The Class InstanceContent.
+ */
+@Immutable
+final public class InstanceContent implements Serializable {
+
+  private static final long serialVersionUID = -8620668863064613841L;
+
+  private long instanceIndex;
+  private int classifierIndex;
+  private int evaluationIndex;
+  private SerializableInstance instance;
+  private boolean isTraining;
+  private boolean isTesting;
+  private boolean isLast = false;
+
+  public InstanceContent() {
+
+  }
+
+  /**
+   * Instantiates a new instance event.
+   *
+   * @param index
+   *          the index
+   * @param instance
+   *          the instance
+   * @param isTraining
+   *          the is training
+   */
+  public InstanceContent(long index, Instance instance,
+      boolean isTraining, boolean isTesting) {
+    if (instance != null) {
+      this.instance = new SerializableInstance(instance);
+    }
+    this.instanceIndex = index;
+    this.isTraining = isTraining;
+    this.isTesting = isTesting;
+  }
+
+  /**
+   * Gets the single instance of InstanceEvent.
+   * 
+   * @return the instance.
+   */
+  public Instance getInstance() {
+    return instance;
+  }
+
+  /**
+   * Gets the instance index.
+   * 
+   * @return the index of the data vector.
+   */
+  public long getInstanceIndex() {
+    return instanceIndex;
+  }
+
+  /**
+   * Gets the class id.
+   * 
+   * @return the true class of the vector.
+   */
+  public int getClassId() {
+    // return classId;
+    return (int) instance.classValue();
+  }
+
+  /**
+   * Checks if is training.
+   * 
+   * @return true if this is training data.
+   */
+  public boolean isTraining() {
+    return isTraining;
+  }
+
+  /**
+   * Set training flag.
+   * 
+   * @param training
+   *          flag.
+   */
+  public void setTraining(boolean training) {
+    this.isTraining = training;
+  }
+
+  /**
+   * Checks if is testing.
+   * 
+   * @return true if this is testing data.
+   */
+  public boolean isTesting() {
+    return isTesting;
+  }
+
+  /**
+   * Set testing flag.
+   * 
+   * @param testing
+   *          flag.
+   */
+  public void setTesting(boolean testing) {
+    this.isTesting = testing;
+  }
+
+  /**
+   * Gets the classifier index.
+   * 
+   * @return the classifier index
+   */
+  public int getClassifierIndex() {
+    return classifierIndex;
+  }
+
+  /**
+   * Sets the classifier index.
+   * 
+   * @param classifierIndex
+   *          the new classifier index
+   */
+  public void setClassifierIndex(int classifierIndex) {
+    this.classifierIndex = classifierIndex;
+  }
+
+  /**
+   * Gets the evaluation index.
+   * 
+   * @return the evaluation index
+   */
+  public int getEvaluationIndex() {
+    return evaluationIndex;
+  }
+
+  /**
+   * Sets the evaluation index.
+   * 
+   * @param evaluationIndex
+   *          the new evaluation index
+   */
+  public void setEvaluationIndex(int evaluationIndex) {
+    this.evaluationIndex = evaluationIndex;
+  }
+
+  /**
+   * Sets the instance index.
+   *
+   * @param instanceIndex
+   *          the new evaluation index
+   */
+  public void setInstanceIndex(long instanceIndex) {
+    this.instanceIndex = instanceIndex;
+  }
+
+  public boolean isLastEvent() {
+    return isLast;
+  }
+
+  public void setLast(boolean isLast) {
+    this.isLast = isLast;
+  }
+
+  @Override
+  public String toString() {
+    return String
+        .format(
+            "InstanceContent [instanceIndex=%s, classifierIndex=%s, evaluationIndex=%s, instance=%s, isTraining=%s, isTesting=%s, isLast=%s]",
+            instanceIndex, classifierIndex, evaluationIndex, instance, isTraining, isTesting, isLast);
+  }
+}