Merge remote-tracking branch 'origin/livedoc' into develop
diff --git a/README.md b/README.md
index 52cf38d..252cb41 100644
--- a/README.md
+++ b/README.md
@@ -78,7 +78,7 @@
* Follow [@predictionio](https://twitter.com/predictionio) on Twitter.
* Read and subscribe to [the
- Newsletter](http://prediction.us6.list-manage1.com/subscribe?u=d8c0435d851c1310fc64c6e26&id=8c6c1b46d0).
+ Newsletter](https://prediction.io/#newsletter).
* Join the [Community
Forum](https://groups.google.com/forum/#!forum/predictionio-user).
diff --git a/bin/install.sh b/bin/install.sh
index f3d0bde..f54aac3 100755
--- a/bin/install.sh
+++ b/bin/install.sh
@@ -71,6 +71,7 @@
echo -e "\033[1;33mForcing Docker defaults!\033[0m"
pio_dir=${PIO_DIR}
vendors_dir=${pio_dir}/vendors
+ source_setup=${ES_HB}
spark_dir=${vendors_dir}/spark-${SPARK_VERSION}
elasticsearch_dir=${vendors_dir}/elasticsearch-${ELASTICSEARCH_VERSION}
diff --git a/build.sbt b/build.sbt
index d39dc89..787ac3b 100644
--- a/build.sbt
+++ b/build.sbt
@@ -18,7 +18,7 @@
name := "pio"
-version in ThisBuild := "0.9.4"
+version in ThisBuild := "0.9.5-SNAPSHOT"
organization in ThisBuild := "io.prediction"
diff --git a/core/src/main/scala/io/prediction/workflow/Workflow.scala b/core/src/main/scala/io/prediction/workflow/Workflow.scala
index 5b3390e..c0543ab 100644
--- a/core/src/main/scala/io/prediction/workflow/Workflow.scala
+++ b/core/src/main/scala/io/prediction/workflow/Workflow.scala
@@ -103,7 +103,7 @@
evaluator: BaseEvaluator[EEI, EQ, EP, EA, ER],
env: Map[String, String] = WorkflowUtils.pioEnvVars,
params: WorkflowParams = WorkflowParams()) {
- runEvaluation(
+ runEvaluationViaCoreWorkflow(
evaluation = evaluation,
engine = engine,
engineParamsList = engineParamsList,
@@ -115,7 +115,7 @@
/** :: Experimental :: */
@Experimental
- def runEvaluation[EI, Q, P, A, R <: BaseEvaluatorResult](
+ def runEvaluationViaCoreWorkflow[EI, Q, P, A, R <: BaseEvaluatorResult](
evaluation: Evaluation,
engine: BaseEngine[EI, Q, P, A],
engineParamsList: Seq[EngineParams],
diff --git a/data/src/main/scala/io/prediction/data/api/EventServer.scala b/data/src/main/scala/io/prediction/data/api/EventServer.scala
index 40bd37a..0ebc279 100644
--- a/data/src/main/scala/io/prediction/data/api/EventServer.scala
+++ b/data/src/main/scala/io/prediction/data/api/EventServer.scala
@@ -30,6 +30,7 @@
import io.prediction.data.storage.DateTimeJson4sSupport
import io.prediction.data.storage.Event
import io.prediction.data.storage.EventJson4sSupport
+import io.prediction.data.storage.BatchEventsJson4sSupport
import io.prediction.data.storage.LEvents
import io.prediction.data.storage.Storage
import org.json4s.DefaultFormats
@@ -46,6 +47,7 @@
import scala.concurrent.ExecutionContext
import scala.concurrent.Future
+import scala.util.{Try, Success, Failure}
class EventServiceActor(
val eventClient: LEvents,
@@ -56,11 +58,14 @@
object Json4sProtocol extends Json4sSupport {
implicit def json4sFormats: Formats = DefaultFormats +
new EventJson4sSupport.APISerializer +
+ new BatchEventsJson4sSupport.APISerializer +
// NOTE: don't use Json4s JodaTimeSerializers since it has issues,
// some format not converted, or timezone not correct
new DateTimeJson4sSupport.Serializer
}
+ val MaxNumberOfEventsPerBatchRequest = 50
+
val log = Logging(context.system, this)
// we use the enclosing ActorContext's or ActorSystem's dispatcher for our
@@ -339,6 +344,73 @@
}
}
} ~
+ path("batch" / "events.json") {
+
+ import Json4sProtocol._
+
+ post {
+ handleExceptions(Common.exceptionHandler) {
+ handleRejections(rejectionHandler) {
+ authenticate(withAccessKey) { authData =>
+ val appId = authData.appId
+ val channelId = authData.channelId
+ val allowedEvents = authData.events
+ val handleEvent: PartialFunction[Try[Event], Future[Map[String, Any]]] = {
+ case Success(event) => {
+ if (allowedEvents.isEmpty || allowedEvents.contains(event.event)) {
+ pluginContext.inputBlockers.values.foreach(
+ _.process(EventInfo(
+ appId = appId,
+ channelId = channelId,
+ event = event), pluginContext))
+ val data = eventClient.futureInsert(event, appId, channelId).map { id =>
+ pluginsActorRef ! EventInfo(
+ appId = appId,
+ channelId = channelId,
+ event = event)
+ val status = StatusCodes.Created
+ val result = Map(
+ "status" -> status.intValue,
+ "eventId" -> s"${id}")
+ if (config.stats) {
+ statsActorRef ! Bookkeeping(appId, status, event)
+ }
+ result
+ }.recover { case exception =>
+ Map(
+ "status" -> StatusCodes.InternalServerError.intValue,
+ "message" -> s"${exception.getMessage()}")
+ }
+ data
+ } else {
+ Future.successful(Map(
+ "status" -> StatusCodes.Forbidden.intValue,
+ "message" -> s"${event.event} events are not allowed"))
+ }
+ }
+ case Failure(exception) => {
+ Future.successful(Map(
+ "status" -> StatusCodes.BadRequest.intValue,
+ "message" -> s"${exception.getMessage()}"))
+ }
+ }
+
+ entity(as[Seq[Try[Event]]]) { events =>
+ complete {
+ if (events.length <= MaxNumberOfEventsPerBatchRequest) {
+ Future.traverse(events)(handleEvent)
+ } else {
+ (StatusCodes.BadRequest,
+ Map("message" -> (s"Batch request must have less than or equal to " +
+ s"${MaxNumberOfEventsPerBatchRequest} events")))
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+ } ~
path("stats.json") {
import Json4sProtocol._
diff --git a/data/src/main/scala/io/prediction/data/storage/EventJson4sSupport.scala b/data/src/main/scala/io/prediction/data/storage/EventJson4sSupport.scala
index a1935d7..22243c2 100644
--- a/data/src/main/scala/io/prediction/data/storage/EventJson4sSupport.scala
+++ b/data/src/main/scala/io/prediction/data/storage/EventJson4sSupport.scala
@@ -19,6 +19,7 @@
import io.prediction.data.{Utils => DataUtils}
import org.joda.time.DateTime
import org.json4s._
+import scala.util.{Try, Success, Failure}
/** :: DeveloperApi ::
* Support library for dealing with [[Event]] and JSON4S
@@ -211,3 +212,25 @@
class APISerializer extends CustomSerializer[Event](format => (
readJson, writeJson))
}
+
+
+@DeveloperApi
+object BatchEventsJson4sSupport {
+ implicit val formats = DefaultFormats
+
+ @DeveloperApi
+ def readJson: PartialFunction[JValue, Seq[Try[Event]]] = {
+ case JArray(events) => {
+ events.map { event =>
+ try {
+ Success(EventJson4sSupport.readJson(event))
+ } catch {
+ case e: Exception => Failure(e)
+ }
+ }
+ }
+ }
+
+ @DeveloperApi
+ class APISerializer extends CustomSerializer[Seq[Try[Event]]](format => (readJson, Map.empty))
+}
diff --git a/data/src/main/scala/io/prediction/data/storage/Storage.scala b/data/src/main/scala/io/prediction/data/storage/Storage.scala
index 4e0e57e..3ad1400 100644
--- a/data/src/main/scala/io/prediction/data/storage/Storage.scala
+++ b/data/src/main/scala/io/prediction/data/storage/Storage.scala
@@ -15,6 +15,8 @@
package io.prediction.data.storage
+import java.lang.reflect.InvocationTargetException
+
import grizzled.slf4j.Logging
import io.prediction.annotation.DeveloperApi
@@ -208,8 +210,8 @@
}
private def getClient(
- clientConfig: StorageClientConfig,
- pkg: String): BaseStorageClient = {
+ clientConfig: StorageClientConfig,
+ pkg: String): BaseStorageClient = {
val className = "io.prediction.data.storage." + pkg + ".StorageClient"
try {
Class.forName(className).getConstructors()(0).newInstance(clientConfig).
@@ -224,6 +226,14 @@
}
}
+ /** Get the StorageClient config data from PIO Framework's environment variables */
+ def getConfig(sourceName: String): Option[StorageClientConfig] = {
+ if (s2cm.contains(sourceName) && s2cm.get(sourceName).nonEmpty
+ && s2cm.get(sourceName).get.nonEmpty) {
+ Some(s2cm.get(sourceName).get.get.config)
+ } else None
+ }
+
private def updateS2CM(k: String, parallel: Boolean, test: Boolean):
Option[ClientMeta] = {
try {
@@ -246,7 +256,7 @@
}
private[prediction]
- def getDataObject[T](repo: String, test: Boolean = false)
+ def getDataObjectFromRepo[T](repo: String, test: Boolean = false)
(implicit tag: TypeTag[T]): T = {
val repoDOMeta = repositoriesToDataObjectMeta(repo)
val repoDOSourceName = repoDOMeta.sourceName
@@ -348,31 +358,31 @@
}
private[prediction] def getMetaDataEngineManifests(): EngineManifests =
- getDataObject[EngineManifests](MetaDataRepository)
+ getDataObjectFromRepo[EngineManifests](MetaDataRepository)
private[prediction] def getMetaDataEngineInstances(): EngineInstances =
- getDataObject[EngineInstances](MetaDataRepository)
+ getDataObjectFromRepo[EngineInstances](MetaDataRepository)
private[prediction] def getMetaDataEvaluationInstances(): EvaluationInstances =
- getDataObject[EvaluationInstances](MetaDataRepository)
+ getDataObjectFromRepo[EvaluationInstances](MetaDataRepository)
private[prediction] def getMetaDataApps(): Apps =
- getDataObject[Apps](MetaDataRepository)
+ getDataObjectFromRepo[Apps](MetaDataRepository)
private[prediction] def getMetaDataAccessKeys(): AccessKeys =
- getDataObject[AccessKeys](MetaDataRepository)
+ getDataObjectFromRepo[AccessKeys](MetaDataRepository)
private[prediction] def getMetaDataChannels(): Channels =
- getDataObject[Channels](MetaDataRepository)
+ getDataObjectFromRepo[Channels](MetaDataRepository)
private[prediction] def getModelDataModels(): Models =
- getDataObject[Models](ModelDataRepository)
+ getDataObjectFromRepo[Models](ModelDataRepository)
/** Obtains a data access object that returns [[Event]] related local data
* structure.
*/
def getLEvents(test: Boolean = false): LEvents =
- getDataObject[LEvents](EventDataRepository, test = test)
+ getDataObjectFromRepo[LEvents](EventDataRepository, test = test)
/** Obtains a data access object that returns [[Event]] related RDD data
* structure.
diff --git a/data/src/main/scala/io/prediction/data/storage/jdbc/JDBCApps.scala b/data/src/main/scala/io/prediction/data/storage/jdbc/JDBCApps.scala
index 0a34aeb..498dbc4 100644
--- a/data/src/main/scala/io/prediction/data/storage/jdbc/JDBCApps.scala
+++ b/data/src/main/scala/io/prediction/data/storage/jdbc/JDBCApps.scala
@@ -41,7 +41,7 @@
"""
} else {
sql"""
- insert into $tableName values(${app.id}, ${app.name}, $app{description}())
+ insert into $tableName values(${app.id}, ${app.name}, ${app.description})
"""
}
Some(q.updateAndReturnGeneratedKey().apply().toInt)
diff --git a/data/src/main/scala/io/prediction/data/storage/jdbc/JDBCLEvents.scala b/data/src/main/scala/io/prediction/data/storage/jdbc/JDBCLEvents.scala
index 2a15e8f..7a6de4e 100644
--- a/data/src/main/scala/io/prediction/data/storage/jdbc/JDBCLEvents.scala
+++ b/data/src/main/scala/io/prediction/data/storage/jdbc/JDBCLEvents.scala
@@ -146,8 +146,8 @@
DB readOnly { implicit session =>
val tableName = sqls.createUnsafely(JDBCUtils.eventTableName(namespace, appId, channelId))
val whereClause = sqls.toAndConditionOpt(
- startTime.map(x => sqls"startTime >= $x"),
- untilTime.map(x => sqls"endTime < $x"),
+ startTime.map(x => sqls"eventTime >= $x"),
+ untilTime.map(x => sqls"eventTime < $x"),
entityType.map(x => sqls"entityType = $x"),
entityId.map(x => sqls"entityId = $x"),
eventNames.map(x =>
diff --git a/data/test.sh b/data/test.sh
index f882e50..7a71b07 100755
--- a/data/test.sh
+++ b/data/test.sh
@@ -442,3 +442,43 @@
checkGET "/events.json?accessKey=$accessKey&untilTime=abc" 400
checkGET "/events.json?accessKey=$accessKey&startTime=2004-12-13T21:39:45.618Z&untilTime=2004-12-15T21:39:45.618Z" 200
+
+# -----
+# batch request
+# ----
+
+# normal request
+testdata='[{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+ "targetEntityType" : "item",
+ "targetEntityId" : "iid",
+ "properties" : {
+ "someProperty" : "value1",
+ "anotherProperty" : "value2"
+ },
+ "eventTime" : "2004-12-13T21:39:45.618Z"
+}]'
+
+checkPOST "/batch/events.json?accessKey=$accessKey" "$testdata" 200
+
+# request with a malformed event (2nd event)
+# the response code is succesful but the error for individual event is reflected in the response's body.
+testdata='[{
+ "event" : "my_event_1",
+ "entityType" : "user",
+ "entityId" : "uid",
+ "eventTime" : "2004-12-13T21:39:45.618Z"
+}, {
+ "eve" : "my_event_2",
+ "entityType" : "user",
+ "entityId" : "uid",
+ "eventTime" : "2015-12-13T21:39:45.618Z"
+}]'
+
+checkPOST "/batch/events.json?accessKey=$accessKey" "$testdata" 200
+
+# request with too many events (more than 50)
+testdata=`cat data/very_long_batch_request.txt`
+checkPOST "/batch/events.json?accessKey=$accessKey" "$testdata" 400
diff --git a/data/very_long_batch_request.txt b/data/very_long_batch_request.txt
new file mode 100644
index 0000000..8fdd7e2
--- /dev/null
+++ b/data/very_long_batch_request.txt
@@ -0,0 +1,205 @@
+[{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+},{
+ "event" : "my_event",
+ "entityType" : "user",
+ "entityId" : "uid",
+}]
\ No newline at end of file
diff --git a/docs/manual/source/demo/textclassification.html.md b/docs/manual/source/demo/textclassification.html.md
index 8246cd0..7f8719b 100644
--- a/docs/manual/source/demo/textclassification.html.md
+++ b/docs/manual/source/demo/textclassification.html.md
@@ -500,7 +500,7 @@
{
"name": "regParam",
"params": {
- "regParam": 2,5
+ "regParam": 0.1
}
}
]
diff --git a/examples/experimental/scala-cleanup-app/README.md b/examples/experimental/scala-cleanup-app/README.md
new file mode 100644
index 0000000..98168dd
--- /dev/null
+++ b/examples/experimental/scala-cleanup-app/README.md
@@ -0,0 +1,12 @@
+# Removing old events from app
+
+## Documentation
+
+This shows how to remove old events from the certain app.
+
+Parameters in engine.json are appId and cutoffTime.
+All events in that appId before the cutoffTime are removed,
+including $set, $unset and $delete
+(so please adapt it for use when you want to preserve these special events).
+
+To use, edit `engine.json`, run `pio build` then `pio train`.
diff --git a/examples/experimental/scala-cleanup-app/build.sbt b/examples/experimental/scala-cleanup-app/build.sbt
new file mode 100644
index 0000000..dc87bee
--- /dev/null
+++ b/examples/experimental/scala-cleanup-app/build.sbt
@@ -0,0 +1,12 @@
+import AssemblyKeys._
+
+assemblySettings
+
+name := "template-scala-parallel-vanilla"
+
+organization := "io.prediction"
+
+libraryDependencies ++= Seq(
+ "io.prediction" %% "core" % "0.9.4-SNAPSHOT" % "provided",
+ "org.apache.spark" %% "spark-core" % "1.3.1" % "provided",
+ "org.apache.spark" %% "spark-mllib" % "1.3.1" % "provided")
diff --git a/examples/experimental/scala-cleanup-app/engine.json b/examples/experimental/scala-cleanup-app/engine.json
new file mode 100644
index 0000000..ab467c9
--- /dev/null
+++ b/examples/experimental/scala-cleanup-app/engine.json
@@ -0,0 +1,11 @@
+{
+ "id": "default",
+ "description": "Default settings",
+ "engineFactory": "io.prediction.examples.experimental.cleanupapp.VanillaEngine",
+ "datasource": {
+ "params" : {
+ "appId": 1000000000,
+ "cutoffTime": "2014-04-29T00:00:00.000Z"
+ }
+ }
+}
diff --git a/examples/experimental/scala-cleanup-app/project/assembly.sbt b/examples/experimental/scala-cleanup-app/project/assembly.sbt
new file mode 100644
index 0000000..54c3252
--- /dev/null
+++ b/examples/experimental/scala-cleanup-app/project/assembly.sbt
@@ -0,0 +1 @@
+addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.11.2")
diff --git a/examples/experimental/scala-cleanup-app/src/main/scala/Algorithm.scala b/examples/experimental/scala-cleanup-app/src/main/scala/Algorithm.scala
new file mode 100644
index 0000000..2b3bbab
--- /dev/null
+++ b/examples/experimental/scala-cleanup-app/src/main/scala/Algorithm.scala
@@ -0,0 +1,32 @@
+package io.prediction.examples.experimental.cleanupapp
+
+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 P2LAlgorithm[TrainingData, Model, Query, PredictedResult] {
+
+ @transient lazy val logger = Logger[this.type]
+
+ def train(sc: SparkContext, data: TrainingData): Model = {
+ new Model
+ }
+
+ def predict(model: Model, query: Query): PredictedResult = {
+ // Prefix the query with the model data
+ PredictedResult(p = "")
+ }
+}
+
+class Model extends Serializable {
+ override def toString = "Model"
+}
diff --git a/examples/experimental/scala-cleanup-app/src/main/scala/DataSource.scala b/examples/experimental/scala-cleanup-app/src/main/scala/DataSource.scala
new file mode 100644
index 0000000..41ce53e
--- /dev/null
+++ b/examples/experimental/scala-cleanup-app/src/main/scala/DataSource.scala
@@ -0,0 +1,80 @@
+package io.prediction.examples.experimental.cleanupapp
+
+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 com.github.nscala_time.time.Imports._
+
+import grizzled.slf4j.Logger
+
+import scala.concurrent.ExecutionContext.Implicits.global
+import scala.concurrent.{Await, Future}
+
+case class DataSourceParams(
+ appId: Int,
+ cutoffTime: DateTime
+) 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 lEventsDb = Storage.getLEvents()
+ logger.info(s"CleanupApp: $dsp")
+
+ val countBefore = eventsDb.find(
+ appId = dsp.appId
+ )(sc).count
+ logger.info(s"Event count before cleanup: $countBefore")
+
+ val countRemove = eventsDb.find(
+ appId = dsp.appId,
+ untilTime = Some(dsp.cutoffTime)
+ )(sc).count
+ logger.info(s"Number of events to remove: $countRemove")
+
+ logger.info(s"Remove events from appId ${dsp.appId}")
+ val eventsToRemove: Array[String] = eventsDb.find(
+ appId = dsp.appId,
+ untilTime = Some(dsp.cutoffTime)
+ )(sc).map { case e =>
+ e.eventId.getOrElse("")
+ }.collect
+
+ var lastFuture: Future[Boolean] = Future[Boolean] {true}
+ eventsToRemove.foreach { case eventId =>
+ if (eventId != "") {
+ lastFuture = lEventsDb.futureDelete(eventId, dsp.appId)
+ }
+ }
+ // No, it's not correct to just wait for the last result.
+ // This program only demonstrates how to remove old events.
+ Await.result(lastFuture, scala.concurrent.duration.Duration(5, "minutes"))
+ logger.info(s"Finish cleaning up events to appId ${dsp.appId}")
+
+ val countAfter = eventsDb.find(
+ appId = dsp.appId
+ )(sc).count
+ logger.info(s"Event count after cleanup: $countAfter")
+
+ throw new StopAfterReadInterruption()
+ }
+}
+
+class TrainingData(
+) extends Serializable {
+ override def toString = ""
+}
diff --git a/examples/experimental/scala-cleanup-app/src/main/scala/Engine.scala b/examples/experimental/scala-cleanup-app/src/main/scala/Engine.scala
new file mode 100644
index 0000000..b7ac9dc
--- /dev/null
+++ b/examples/experimental/scala-cleanup-app/src/main/scala/Engine.scala
@@ -0,0 +1,19 @@
+package io.prediction.examples.experimental.cleanupapp
+
+import io.prediction.controller.IEngineFactory
+import io.prediction.controller.Engine
+import io.prediction.controller._
+
+case class Query(q: String) extends Serializable
+
+case class PredictedResult(p: String) extends Serializable
+
+object VanillaEngine extends IEngineFactory {
+ def apply() = {
+ new Engine(
+ classOf[DataSource],
+ PIdentityPreparator(classOf[DataSource]),
+ Map("" -> classOf[Algorithm]),
+ classOf[Serving])
+ }
+}
diff --git a/examples/experimental/scala-cleanup-app/src/main/scala/Preparator.scala b/examples/experimental/scala-cleanup-app/src/main/scala/Preparator.scala
new file mode 100644
index 0000000..54bba9f
--- /dev/null
+++ b/examples/experimental/scala-cleanup-app/src/main/scala/Preparator.scala
@@ -0,0 +1,22 @@
+package io.prediction.examples.experimental.cleanupapp
+
+import io.prediction.controller.PPreparator
+import io.prediction.data.storage.Event
+
+import org.apache.spark.SparkContext
+import org.apache.spark.SparkContext._
+import org.apache.spark.rdd.RDD
+
+/*
+class Preparator
+ extends PPreparator[TrainingData, PreparedData] {
+
+ def prepare(sc: SparkContext, trainingData: TrainingData): PreparedData = {
+ new PreparedData(events = trainingData.events)
+ }
+}
+
+class PreparedData(
+ val events: RDD[Event]
+) extends Serializable
+*/
diff --git a/examples/experimental/scala-cleanup-app/src/main/scala/Serving.scala b/examples/experimental/scala-cleanup-app/src/main/scala/Serving.scala
new file mode 100644
index 0000000..9898307
--- /dev/null
+++ b/examples/experimental/scala-cleanup-app/src/main/scala/Serving.scala
@@ -0,0 +1,13 @@
+package io.prediction.examples.experimental.cleanupapp
+
+import io.prediction.controller.LServing
+
+class Serving
+ extends LServing[Query, PredictedResult] {
+
+ override
+ def serve(query: Query,
+ predictedResults: Seq[PredictedResult]): PredictedResult = {
+ predictedResults.head
+ }
+}
\ No newline at end of file
diff --git a/manifest.json b/manifest.json
index 65911fb..1423946 100644
--- a/manifest.json
+++ b/manifest.json
@@ -1 +1 @@
-{"id":"AQUacEPKfAHhmU7wmxHErpEKiHwCJ1X0","version":"1d33d77ca11aaf0658d2f32e95d881007b406818","name":"PredictionIO","description":"pio-autogen-manifest","files":[],"engineFactory":""}
\ No newline at end of file
+{"id":"ipNrDsdxAtsDjZHRSAqfgOTL0PztjuR3","version":"b6a10c9f995426f7ee739156066bc26b577f2f09","name":"pio","description":"pio-autogen-manifest","files":[],"engineFactory":""}
\ No newline at end of file
diff --git a/sbt/sbt-launch-lib.bash b/sbt/sbt-launch-lib.bash
index 09f151e..10eb7fe 100644
--- a/sbt/sbt-launch-lib.bash
+++ b/sbt/sbt-launch-lib.bash
@@ -37,8 +37,8 @@
acquire_sbt_jar () {
SBT_VERSION=`awk -F "=" '/sbt\\.version/ {print $2}' $sbt_bin_dir/../project/build.properties`
- URL1=http://typesafe.artifactoryonline.com/typesafe/ivy-releases/org.scala-sbt/sbt-launch/${SBT_VERSION}/sbt-launch.jar
- URL2=http://repo.typesafe.com/typesafe/ivy-releases/org.scala-sbt/sbt-launch/${SBT_VERSION}/sbt-launch.jar
+ URL1=http://repo.typesafe.com/typesafe/ivy-releases/org.scala-sbt/sbt-launch/${SBT_VERSION}/sbt-launch.jar
+ URL2=http://typesafe.artifactoryonline.com/typesafe/ivy-releases/org.scala-sbt/sbt-launch/${SBT_VERSION}/sbt-launch.jar
JAR=$sbt_bin_dir/sbt-launch-${SBT_VERSION}.jar
sbt_jar=$JAR
@@ -50,9 +50,9 @@
printf "Attempting to fetch sbt\n"
JAR_DL=${JAR}.part
if hash curl 2>/dev/null; then
- (curl --progress-bar ${URL1} > ${JAR_DL} || curl --progress-bar ${URL2} > ${JAR_DL}) && mv ${JAR_DL} ${JAR}
+ (curl -L --verbose --progress-bar ${URL1} > ${JAR_DL} || curl -L --verbose --progress-bar ${URL2} > ${JAR_DL}) && mv ${JAR_DL} ${JAR}
elif hash wget 2>/dev/null; then
- (wget --progress=bar ${URL1} -O ${JAR_DL} || wget --progress=bar ${URL2} -O ${JAR_DL}) && mv ${JAR_DL} ${JAR}
+ (wget --verbose --progress=bar ${URL1} -O ${JAR_DL} || wget --verbose --progress=bar ${URL2} -O ${JAR_DL}) && mv ${JAR_DL} ${JAR}
else
printf "You do not have curl or wget installed, please install sbt manually from http://www.scala-sbt.org/\n"
exit -1
diff --git a/tools/build.sbt b/tools/build.sbt
index 4e330bb..865386f 100644
--- a/tools/build.sbt
+++ b/tools/build.sbt
@@ -22,7 +22,7 @@
"com.github.scopt" %% "scopt" % "3.2.0",
"io.spray" %% "spray-can" % "1.3.2",
"io.spray" %% "spray-routing" % "1.3.2",
- "me.lessis" %% "semverfi" % "0.1.3",
+ "me.lessis" % "semverfi_2.10" % "0.1.3",
"org.apache.hadoop" % "hadoop-common" % "2.5.0",
"org.apache.hadoop" % "hadoop-hdfs" % "2.5.0",
"org.apache.spark" %% "spark-core" % sparkVersion.value % "provided",
