First commit

Author: Justin Yip

.gitignore

@@ -0,0 +1,4 @@
+manifest.json
+target/
+pio.log
+/pio.sbt

README.md

@@ -0,0 +1,24 @@
+# Event Distribution Checker
+
+It prints out the landscape of an appid.
+
+## To run
+
+edit engine.json. 2 key fields.
+
+- appId: the appId
+- sample: the # of event to be printed for each key. can be empty.
+
+```
+$ pio build
+$ pio train
+...
+[INFO] [BasicChecker] Event (Time, Name, EntityType, TargetEntityType) Distribution
+[INFO] [AggEvent$] [1 x 4]
+                               Count   Freq CumCount CumFreq 
+                            -------- ------ -------- ------- 
+(2015-03,$set,user,None) -> 153.0000 1.0000 153.0000  1.0000 
+...
+[INFO] [CoreWorkflow$] Training interrupted by io.prediction.workflow.StopAfterReadInterruption.
+```
+

build.sbt

@@ -0,0 +1,21 @@
+import AssemblyKeys._
+
+assemblySettings
+
+name := "template-scala-parallel-vanilla"
+
+organization := "io.prediction"
+
+libraryDependencies ++= Seq(
+  "io.prediction"    %% "core"          % pioVersion.value % "provided",
+  "org.apache.spark" %% "spark-core"    % "1.2.0" % "provided",
+  "org.apache.spark" %% "spark-mllib"   % "1.2.0" % "provided",
+  "org.scala-saddle" %% "saddle-core" % "1.3.+")
+
+resolvers ++= Seq(
+  "Sonatype Snapshots" at 
+    "http://oss.sonatype.org/content/repositories/snapshots",
+  "Sonatype Releases" at
+    "http://oss.sonatype.org/content/repositories/releases"
+)
+

engine.json

@@ -0,0 +1,11 @@
+{
+  "id": "default",
+  "description": "Default settings",
+  "engineFactory": "io.prediction.e2.eventdistributionchecker.VanillaEngine",
+  "datasource": {
+    "params" : {
+      "appId": 11,
+      "sample": 0
+    }
+  }
+}

src/main/scala/Algorithm.scala

@@ -0,0 +1,38 @@
+package io.prediction.e2.eventdistributionchecker
+
+import io.prediction.controller.P2LAlgorithm
+import io.prediction.controller.Params
+
+import org.apache.spark.SparkContext
+import org.apache.spark.SparkContext._
+import org.apache.spark.rdd.RDD
+
+import grizzled.slf4j.Logger
+
+//case class AlgorithmParams(mult: Int) extends Params
+
+//class Algorithm(val ap: AlgorithmParams)
+class Algorithm
+  // extends PAlgorithm if Model contains RDD[]
+  extends P2LAlgorithm[PreparedData, Model, Query, PredictedResult] {
+
+  @transient lazy val logger = Logger[this.type]
+
+  def train(sc: SparkContext, data: PreparedData): Model = {
+    // Simply count number of events
+    // and multiple it by the algorithm parameter
+    // and store the number as model
+    val count = data.events.count().toInt
+    new Model(mc = count)
+  }
+
+  def predict(model: Model, query: Query): PredictedResult = {
+    // Prefix the query with the model data
+    val result = s"${model.mc}-${query.q}"
+    PredictedResult(p = result)
+  }
+}
+
+class Model(val mc: Int) extends Serializable {
+  override def toString = s"mc=${mc}"
+}

src/main/scala/DataSource.scala

@@ -0,0 +1,128 @@
+package io.prediction.e2.eventdistributionchecker
+
+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 io.prediction.workflow.StopAfterReadInterruption
+
+import org.apache.spark.SparkContext
+import org.apache.spark.SparkContext._
+import org.apache.spark.rdd.RDD
+import org.apache.spark.FutureAction
+import org.apache.spark.rdd.AsyncRDDActions
+import org.apache.spark.storage.StorageLevel.MEMORY_ONLY_SER
+import org.apache.spark.storage.StorageLevel.MEMORY_ONLY
+import org.apache.spark.Accumulator
+import org.apache.spark.Accumulable
+
+import grizzled.slf4j.Logger
+import scala.concurrent.ExecutionContext.Implicits.global
+import scala.concurrent.Await
+import scala.concurrent.duration._
+
+import com.github.nscala_time.time.Imports._
+import org.apache.spark.mllib.linalg.Vector
+import org.apache.spark.mllib.stat.{MultivariateStatisticalSummary, Statistics}
+import org.apache.spark.mllib.linalg.Vectors
+import org.joda.time.Days
+import org.joda.time.Hours
+
+
+case class DataSourceParams(
+    appId: Int,
+    sample: Option[Int]
+) extends Params
+
+
+abstract class AbstractChecker extends Serializable {
+  def check(): Unit
+} 
+
+class BasicChecker(ds: DataSource, eventsRDD: RDD[Event], sample: Option[Int])
+    extends AbstractChecker {
+  @transient lazy val logger = Logger[this.type]
+
+  import io.prediction.e2.eventdistributionchecker.AggEvent.rddToEventRDD
+  import io.prediction.e2.eventdistributionchecker.AggEvent.rddToKeyedEventRDD
+  
+  val _sample = sample.getOrElse(0)
+  
+  val fTypeDist: FutureAction[Seq[((String, Option[String]), AggEvent.Result)]] =
+  eventsRDD
+  .map(e => ((e.entityType, e.targetEntityType), e))
+  .countAndSample(_sample)
+  .collectAsync()
+
+  val fEventTimeDist = eventsRDD
+    .map{ e => (e.eventTime.toString("yyyy-MM"), e) }
+    .countAndSample(_sample)
+    .collectAsync()
+
+  val fEventTimeNameDist = eventsRDD
+    .map{ e => ((e.eventTime.toString("yyyy-MM"), e.event), e) }
+    .countAndSample(_sample)
+    .collectAsync()
+
+  val fEventTimeNameTypeDist = eventsRDD
+    .map{ e => 
+      ((e.eventTime.toString("yyyy-MM"), e.event, e.entityType, e.targetEntityType), e)
+    }
+    .countAndSample(_sample)
+    .collectAsync()
+  
+  val fTypeDistinctId = EventUtils.distinctEntityId(
+    eventsRDD, _.entityType).collectAsync()
+
+  def check(): Unit = {
+    logger.info("Entity Type Distribution")
+    AggEvent.print(fTypeDist.get(), true)
+
+
+    logger.info("Event Time Distribution")
+    AggEvent.print(fEventTimeDist.get(), true)
+
+    logger.info("Event (Time, Name, EntityType, TargetEntityType) Distribution")
+    AggEvent.print(fEventTimeNameTypeDist.get(), true)
+    
+    logger.info("Event (EntityType) Distinct Id")
+    val typeDistinctIdStats = EventUtils.distStats(fTypeDistinctId.get(), true)
+    logger.info(typeDistinctIdStats)
+  }
+}
+
+
+class DataSource(val dsp: DataSourceParams)
+  extends PDataSource[TrainingData,
+      EmptyEvaluationInfo, Query, EmptyActualResult] {
+
+  @transient lazy val logger = Logger[this.type]
+  @transient lazy val eventsDb = Storage.getPEvents()
+
+  override
+  def readTraining(sc: SparkContext): TrainingData = {
+    val eventsRDD: RDD[Event] = eventsDb.find(appId = dsp.appId)(sc)
+      .setName("EventsRDD")
+      .cache
+
+    logger.info(s"EventsCount: ${eventsRDD.count}")
+
+    val checkers = Seq[AbstractChecker](
+      new BasicChecker(this, eventsRDD, sample=dsp.sample)
+    )
+
+    checkers.foreach(_.check)
+
+    throw new StopAfterReadInterruption()
+  }
+}
+
+class TrainingData(
+  val events: RDD[Event]
+) extends Serializable {
+  override def toString = {
+    s"events: [${events.count()}] (${events.take(2).toList}...)"
+  }
+}

src/main/scala/Engine.scala

@@ -0,0 +1,18 @@
+package io.prediction.e2.eventdistributionchecker
+
+import io.prediction.controller.IEngineFactory
+import io.prediction.controller.Engine
+
+case class Query(q: String) extends Serializable
+
+case class PredictedResult(p: String) extends Serializable
+
+object VanillaEngine extends IEngineFactory {
+  def apply() = {
+    new Engine(
+      classOf[DataSource],
+      classOf[Preparator],
+      Map("" -> classOf[Algorithm]),
+      classOf[Serving])
+  }
+}