Emigration from core repo
diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..c89ee97
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,2 @@
+data/*.txt
+manifest.json
diff --git a/build.sbt b/build.sbt
new file mode 100644
index 0000000..6105185
--- /dev/null
+++ b/build.sbt
@@ -0,0 +1,14 @@
+import AssemblyKeys._
+
+assemblySettings
+
+name := "template-scala-parallel-classification"
+
+organization := "io.prediction"
+
+libraryDependencies ++= Seq(
+ "io.prediction" %% "core" % "0.8.6-SNAPSHOT" % "provided",
+ "commons-io" % "commons-io" % "2.4",
+ "org.apache.spark" %% "spark-core" % "1.2.0" % "provided",
+ "org.apache.spark" %% "spark-mllib" % "1.2.0" % "provided",
+ "org.json4s" %% "json4s-native" % "3.2.10")
diff --git a/data/import_eventserver.py b/data/import_eventserver.py
new file mode 100644
index 0000000..a35235d
--- /dev/null
+++ b/data/import_eventserver.py
@@ -0,0 +1,46 @@
+"""
+Import sample data for classification engine
+"""
+
+import predictionio
+import argparse
+
+def import_events(client, file):
+ f = open(file, 'r')
+ count = 0
+ print "Importing data..."
+ for line in f:
+ data = line.rstrip('\r\n').split(",")
+ plan = data[0]
+ attr = data[1].split(" ")
+ client.create_event(
+ event="$set",
+ entity_type="user",
+ entity_id=str(count), # use the count num as user ID
+ properties= {
+ "attr0" : int(attr[0]),
+ "attr1" : int(attr[1]),
+ "attr2" : int(attr[2]),
+ "plan" : int(plan)
+ }
+ )
+ count += 1
+ f.close()
+ print "%s events are imported." % count
+
+if __name__ == '__main__':
+ parser = argparse.ArgumentParser(
+ description="Import sample data for classification engine")
+ parser.add_argument('--access_key', default='invald_access_key')
+ parser.add_argument('--url', default="http://localhost:7070")
+ parser.add_argument('--file', default="./data/sample_naive_bayes_data.txt")
+
+ args = parser.parse_args()
+ print args
+
+ client = predictionio.EventClient(
+ access_key=args.access_key,
+ url=args.url,
+ threads=5,
+ qsize=500)
+ import_events(client, args.file)
diff --git a/engine.json b/engine.json
new file mode 100644
index 0000000..d20a2ad
--- /dev/null
+++ b/engine.json
@@ -0,0 +1,18 @@
+{
+ "id": "default",
+ "description": "Default settings",
+ "engineFactory": "org.template.classification.ClassificationEngine",
+ "datasource": {
+ "params": {
+ "appId": 2
+ }
+ },
+ "algorithms": [
+ {
+ "name": "naive",
+ "params": {
+ "lambda": 1.0
+ }
+ }
+ ]
+}
diff --git a/project/assembly.sbt b/project/assembly.sbt
new file mode 100644
index 0000000..54c3252
--- /dev/null
+++ b/project/assembly.sbt
@@ -0,0 +1 @@
+addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.11.2")
diff --git a/src/main/scala/DataSource.scala b/src/main/scala/DataSource.scala
new file mode 100644
index 0000000..c2a422d
--- /dev/null
+++ b/src/main/scala/DataSource.scala
@@ -0,0 +1,60 @@
+package org.template.classification
+
+import io.prediction.controller.PDataSource
+import io.prediction.controller.EmptyEvaluationInfo
+import io.prediction.controller.EmptyActualResult
+import io.prediction.controller.Params
+import io.prediction.data.storage.Event
+import io.prediction.data.storage.Storage
+
+import org.apache.spark.SparkContext
+import org.apache.spark.SparkContext._
+import org.apache.spark.rdd.RDD
+import org.apache.spark.mllib.regression.LabeledPoint
+import org.apache.spark.mllib.linalg.Vectors
+
+import grizzled.slf4j.Logger
+
+case class DataSourceParams(appId: Int) extends Params
+
+class DataSource(val dsp: DataSourceParams)
+ extends PDataSource[TrainingData,
+ EmptyEvaluationInfo, Query, EmptyActualResult] {
+
+ @transient lazy val logger = Logger[this.type]
+
+ override
+ def readTraining(sc: SparkContext): TrainingData = {
+ val eventsDb = Storage.getPEvents()
+ val labeledPoints: RDD[LabeledPoint] = eventsDb.aggregateProperties(
+ appId = dsp.appId,
+ entityType = "user",
+ // only keep entities with these required properties defined
+ required = Some(List("plan", "attr0", "attr1", "attr2")))(sc)
+ // aggregateProperties() returns RDD pair of
+ // entity ID and its aggregated properties
+ .map { case (entityId, properties) =>
+ try {
+ LabeledPoint(properties.get[Double]("plan"),
+ Vectors.dense(Array(
+ properties.get[Double]("attr0"),
+ properties.get[Double]("attr1"),
+ properties.get[Double]("attr2")
+ ))
+ )
+ } catch {
+ case e: Exception => {
+ logger.error(s"Failed to get properties ${properties} of" +
+ s" ${entityId}. Exception: ${e}.")
+ throw e
+ }
+ }
+ }
+
+ new TrainingData(labeledPoints)
+ }
+}
+
+class TrainingData(
+ val labeledPoints: RDD[LabeledPoint]
+) extends Serializable
diff --git a/src/main/scala/Engine.scala b/src/main/scala/Engine.scala
new file mode 100644
index 0000000..b2eb05e
--- /dev/null
+++ b/src/main/scala/Engine.scala
@@ -0,0 +1,22 @@
+package org.template.classification
+
+import io.prediction.controller.IEngineFactory
+import io.prediction.controller.Engine
+
+class Query(
+ val features: Array[Double]
+) extends Serializable
+
+class PredictedResult(
+ val label: Double
+) extends Serializable
+
+object ClassificationEngine extends IEngineFactory {
+ def apply() = {
+ new Engine(
+ classOf[DataSource],
+ classOf[Preparator],
+ Map("naive" -> classOf[NaiveBayesAlgorithm]),
+ classOf[Serving])
+ }
+}
diff --git a/src/main/scala/NaiveBayesAlgorithm.scala b/src/main/scala/NaiveBayesAlgorithm.scala
new file mode 100644
index 0000000..f4e57a7
--- /dev/null
+++ b/src/main/scala/NaiveBayesAlgorithm.scala
@@ -0,0 +1,37 @@
+package org.template.classification
+
+import io.prediction.controller.P2LAlgorithm
+import io.prediction.controller.Params
+
+import org.apache.spark.mllib.classification.NaiveBayes
+import org.apache.spark.mllib.classification.NaiveBayesModel
+import org.apache.spark.mllib.linalg.Vectors
+
+import grizzled.slf4j.Logger
+
+case class AlgorithmParams(
+ lambda: Double
+) extends Params
+
+// extends P2LAlgorithm because the MLlib's NaiveBayesModel doesn't contain RDD.
+class NaiveBayesAlgorithm(val ap: AlgorithmParams)
+ extends P2LAlgorithm[PreparedData, NaiveBayesModel, Query, PredictedResult] {
+
+ @transient lazy val logger = Logger[this.type]
+
+ def train(data: PreparedData): NaiveBayesModel = {
+ // MLLib NaiveBayes cannot handle empty training data.
+ require(!data.labeledPoints.take(1).isEmpty,
+ s"RDD[labeldPoints] in PreparedData cannot be empty." +
+ " Please check if DataSource generates TrainingData" +
+ " and Preprator generates PreparedData correctly.")
+
+ NaiveBayes.train(data.labeledPoints, ap.lambda)
+ }
+
+ def predict(model: NaiveBayesModel, query: Query): PredictedResult = {
+ val label = model.predict(Vectors.dense(query.features))
+ new PredictedResult(label)
+ }
+
+}
diff --git a/src/main/scala/Preparator.scala b/src/main/scala/Preparator.scala
new file mode 100644
index 0000000..4e581ea
--- /dev/null
+++ b/src/main/scala/Preparator.scala
@@ -0,0 +1,19 @@
+package org.template.classification
+
+import io.prediction.controller.PPreparator
+
+import org.apache.spark.SparkContext
+import org.apache.spark.SparkContext._
+import org.apache.spark.rdd.RDD
+import org.apache.spark.mllib.regression.LabeledPoint
+
+class PreparedData(
+ val labeledPoints: RDD[LabeledPoint]
+) extends Serializable
+
+class Preparator extends PPreparator[TrainingData, PreparedData] {
+
+ def prepare(sc: SparkContext, trainingData: TrainingData): PreparedData = {
+ new PreparedData(trainingData.labeledPoints)
+ }
+}
diff --git a/src/main/scala/Serving.scala b/src/main/scala/Serving.scala
new file mode 100644
index 0000000..ef06088
--- /dev/null
+++ b/src/main/scala/Serving.scala
@@ -0,0 +1,12 @@
+package org.template.classification
+
+import io.prediction.controller.LServing
+
+class Serving extends LServing[Query, PredictedResult] {
+
+ override
+ def serve(query: Query,
+ predictedResults: Seq[PredictedResult]): PredictedResult = {
+ predictedResults.head
+ }
+}
