You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@predictionio.apache.org by do...@apache.org on 2016/07/18 20:30:07 UTC
[3/3] incubator-predictionio git commit: Merge branch
'ziemin/disable_upgrade' into develop
Merge branch 'ziemin/disable_upgrade' into develop
Project: http://git-wip-us.apache.org/repos/asf/incubator-predictionio/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-predictionio/commit/c508c791
Tree: http://git-wip-us.apache.org/repos/asf/incubator-predictionio/tree/c508c791
Diff: http://git-wip-us.apache.org/repos/asf/incubator-predictionio/diff/c508c791
Branch: refs/heads/develop
Commit: c508c791d85061bc053fbffc704ac953be814234
Parents: 02a5655 3b87275
Author: Donald Szeto <do...@apache.org>
Authored: Mon Jul 18 13:28:06 2016 -0700
Committer: Donald Szeto <do...@apache.org>
Committed: Mon Jul 18 13:28:06 2016 -0700
----------------------------------------------------------------------
.../predictionio/workflow/CoreWorkflow.scala | 3 --
.../predictionio/workflow/CreateServer.scala | 18 -----------
.../predictionio/workflow/WorkflowUtils.scala | 32 --------------------
.../predictionio/tools/console/Console.scala | 9 +-----
4 files changed, 1 insertion(+), 61 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/c508c791/core/src/main/scala/org/apache/predictionio/workflow/CoreWorkflow.scala
----------------------------------------------------------------------
diff --cc core/src/main/scala/org/apache/predictionio/workflow/CoreWorkflow.scala
index 6a27e87,0000000..513b22b
mode 100644,000000..100644
--- a/core/src/main/scala/org/apache/predictionio/workflow/CoreWorkflow.scala
+++ b/core/src/main/scala/org/apache/predictionio/workflow/CoreWorkflow.scala
@@@ -1,163 -1,0 +1,160 @@@
+/** Copyright 2015 TappingStone, Inc.
+ *
+ * Licensed 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.
+ */
+
+package org.apache.predictionio.workflow
+
+import org.apache.predictionio.controller.EngineParams
+import org.apache.predictionio.controller.Evaluation
+import org.apache.predictionio.core.BaseEngine
+import org.apache.predictionio.core.BaseEvaluator
+import org.apache.predictionio.core.BaseEvaluatorResult
+import org.apache.predictionio.data.storage.EngineInstance
+import org.apache.predictionio.data.storage.EvaluationInstance
+import org.apache.predictionio.data.storage.Model
+import org.apache.predictionio.data.storage.Storage
+
+import com.github.nscala_time.time.Imports.DateTime
+import grizzled.slf4j.Logger
+
+import scala.language.existentials
+
+/** CoreWorkflow handles PredictionIO metadata and environment variables of
+ * training and evaluation.
+ */
+object CoreWorkflow {
+ @transient lazy val logger = Logger[this.type]
+ @transient lazy val engineInstances = Storage.getMetaDataEngineInstances
+ @transient lazy val evaluationInstances =
+ Storage.getMetaDataEvaluationInstances()
+
+ def runTrain[EI, Q, P, A](
+ engine: BaseEngine[EI, Q, P, A],
+ engineParams: EngineParams,
+ engineInstance: EngineInstance,
+ env: Map[String, String] = WorkflowUtils.pioEnvVars,
+ params: WorkflowParams = WorkflowParams()) {
+ logger.debug("Starting SparkContext")
+ val mode = "training"
- WorkflowUtils.checkUpgrade(mode, engineInstance.engineFactory)
+
+ val batch = if (params.batch.nonEmpty) {
+ s"{engineInstance.engineFactory} (${params.batch}})"
+ } else {
+ engineInstance.engineFactory
+ }
+ val sc = WorkflowContext(
+ batch,
+ env,
+ params.sparkEnv,
+ mode.capitalize)
+
+ try {
+
+ val models: Seq[Any] = engine.train(
+ sc = sc,
+ engineParams = engineParams,
+ engineInstanceId = engineInstance.id,
+ params = params
+ )
+
+ val instanceId = Storage.getMetaDataEngineInstances
+
+ val kryo = KryoInstantiator.newKryoInjection
+
+ logger.info("Inserting persistent model")
+ Storage.getModelDataModels.insert(Model(
+ id = engineInstance.id,
+ models = kryo(models)))
+
+ logger.info("Updating engine instance")
+ val engineInstances = Storage.getMetaDataEngineInstances
+ engineInstances.update(engineInstance.copy(
+ status = "COMPLETED",
+ endTime = DateTime.now
+ ))
+
+ logger.info("Training completed successfully.")
+ } catch {
+ case e @(
+ _: StopAfterReadInterruption |
+ _: StopAfterPrepareInterruption) => {
+ logger.info(s"Training interrupted by $e.")
+ }
+ } finally {
+ logger.debug("Stopping SparkContext")
+ sc.stop()
+ }
+ }
+
+ def runEvaluation[EI, Q, P, A, R <: BaseEvaluatorResult](
+ evaluation: Evaluation,
+ engine: BaseEngine[EI, Q, P, A],
+ engineParamsList: Seq[EngineParams],
+ evaluationInstance: EvaluationInstance,
+ evaluator: BaseEvaluator[EI, Q, P, A, R],
+ env: Map[String, String] = WorkflowUtils.pioEnvVars,
+ params: WorkflowParams = WorkflowParams()) {
+ logger.info("runEvaluation started")
+ logger.debug("Start SparkContext")
+
+ val mode = "evaluation"
+
- WorkflowUtils.checkUpgrade(mode, engine.getClass.getName)
-
+ val batch = if (params.batch.nonEmpty) {
+ s"{evaluation.getClass.getName} (${params.batch}})"
+ } else {
+ evaluation.getClass.getName
+ }
+ val sc = WorkflowContext(
+ batch,
+ env,
+ params.sparkEnv,
+ mode.capitalize)
+ val evaluationInstanceId = evaluationInstances.insert(evaluationInstance)
+
+ logger.info(s"Starting evaluation instance ID: $evaluationInstanceId")
+
+ val evaluatorResult: BaseEvaluatorResult = EvaluationWorkflow.runEvaluation(
+ sc,
+ evaluation,
+ engine,
+ engineParamsList,
+ evaluator,
+ params)
+
+ if (evaluatorResult.noSave) {
+ logger.info(s"This evaluation result is not inserted into database: $evaluatorResult")
+ } else {
+ val evaluatedEvaluationInstance = evaluationInstance.copy(
+ status = "EVALCOMPLETED",
+ id = evaluationInstanceId,
+ endTime = DateTime.now,
+ evaluatorResults = evaluatorResult.toOneLiner,
+ evaluatorResultsHTML = evaluatorResult.toHTML,
+ evaluatorResultsJSON = evaluatorResult.toJSON
+ )
+
+ logger.info(s"Updating evaluation instance with result: $evaluatorResult")
+
+ evaluationInstances.update(evaluatedEvaluationInstance)
+ }
+
+ logger.debug("Stop SparkContext")
+
+ sc.stop()
+
+ logger.info("runEvaluation completed")
+ }
+}
+
+
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/c508c791/core/src/main/scala/org/apache/predictionio/workflow/CreateServer.scala
----------------------------------------------------------------------
diff --cc core/src/main/scala/org/apache/predictionio/workflow/CreateServer.scala
index d4f6323,0000000..9e12b35
mode 100644,000000..100644
--- a/core/src/main/scala/org/apache/predictionio/workflow/CreateServer.scala
+++ b/core/src/main/scala/org/apache/predictionio/workflow/CreateServer.scala
@@@ -1,737 -1,0 +1,719 @@@
+/** Copyright 2015 TappingStone, Inc.
+ *
+ * Licensed 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.
+ */
+
+package org.apache.predictionio.workflow
+
+import java.io.PrintWriter
+import java.io.Serializable
+import java.io.StringWriter
+import java.util.concurrent.TimeUnit
+
+import akka.actor._
+import akka.event.Logging
+import akka.io.IO
+import akka.pattern.ask
+import akka.util.Timeout
+import com.github.nscala_time.time.Imports.DateTime
+import com.twitter.bijection.Injection
+import com.twitter.chill.KryoBase
+import com.twitter.chill.KryoInjection
+import com.twitter.chill.ScalaKryoInstantiator
+import com.typesafe.config.ConfigFactory
+import de.javakaffee.kryoserializers.SynchronizedCollectionsSerializer
+import grizzled.slf4j.Logging
+import org.apache.predictionio.authentication.KeyAuthentication
+import org.apache.predictionio.configuration.SSLConfiguration
+import org.apache.predictionio.controller.Engine
+import org.apache.predictionio.controller.Params
+import org.apache.predictionio.controller.Utils
+import org.apache.predictionio.controller.WithPrId
+import org.apache.predictionio.core.BaseAlgorithm
+import org.apache.predictionio.core.BaseServing
+import org.apache.predictionio.core.Doer
+import org.apache.predictionio.data.storage.EngineInstance
+import org.apache.predictionio.data.storage.EngineManifest
+import org.apache.predictionio.data.storage.Storage
+import org.apache.predictionio.workflow.JsonExtractorOption.JsonExtractorOption
+import org.json4s._
+import org.json4s.native.JsonMethods._
+import org.json4s.native.Serialization.write
+import spray.can.Http
+import spray.can.server.ServerSettings
+import spray.http.MediaTypes._
+import spray.http._
+import spray.httpx.Json4sSupport
+import spray.routing._
+import spray.routing.authentication.{UserPass, BasicAuth}
+
+import scala.concurrent.ExecutionContext.Implicits.global
+import scala.concurrent.Future
+import scala.concurrent.duration._
+import scala.concurrent.future
+import scala.language.existentials
+import scala.util.Failure
+import scala.util.Random
+import scala.util.Success
+import scalaj.http.HttpOptions
+
+class KryoInstantiator(classLoader: ClassLoader) extends ScalaKryoInstantiator {
+ override def newKryo(): KryoBase = {
+ val kryo = super.newKryo()
+ kryo.setClassLoader(classLoader)
+ SynchronizedCollectionsSerializer.registerSerializers(kryo)
+ kryo
+ }
+}
+
+object KryoInstantiator extends Serializable {
+ def newKryoInjection : Injection[Any, Array[Byte]] = {
+ val kryoInstantiator = new KryoInstantiator(getClass.getClassLoader)
+ KryoInjection.instance(kryoInstantiator)
+ }
+}
+
+case class ServerConfig(
+ batch: String = "",
+ engineInstanceId: String = "",
+ engineId: Option[String] = None,
+ engineVersion: Option[String] = None,
+ engineVariant: String = "",
+ env: Option[String] = None,
+ ip: String = "0.0.0.0",
+ port: Int = 8000,
+ feedback: Boolean = false,
+ eventServerIp: String = "0.0.0.0",
+ eventServerPort: Int = 7070,
+ accessKey: Option[String] = None,
+ logUrl: Option[String] = None,
+ logPrefix: Option[String] = None,
+ logFile: Option[String] = None,
+ verbose: Boolean = false,
+ debug: Boolean = false,
+ jsonExtractor: JsonExtractorOption = JsonExtractorOption.Both)
+
+case class StartServer()
+case class BindServer()
+case class StopServer()
+case class ReloadServer()
- case class UpgradeCheck()
+
+
+object CreateServer extends Logging {
+ val actorSystem = ActorSystem("pio-server")
+ val engineInstances = Storage.getMetaDataEngineInstances
+ val engineManifests = Storage.getMetaDataEngineManifests
+ val modeldata = Storage.getModelDataModels
+
+ def main(args: Array[String]): Unit = {
+ val parser = new scopt.OptionParser[ServerConfig]("CreateServer") {
+ opt[String]("batch") action { (x, c) =>
+ c.copy(batch = x)
+ } text("Batch label of the deployment.")
+ opt[String]("engineId") action { (x, c) =>
+ c.copy(engineId = Some(x))
+ } text("Engine ID.")
+ opt[String]("engineVersion") action { (x, c) =>
+ c.copy(engineVersion = Some(x))
+ } text("Engine version.")
+ opt[String]("engine-variant") required() action { (x, c) =>
+ c.copy(engineVariant = x)
+ } text("Engine variant JSON.")
+ opt[String]("ip") action { (x, c) =>
+ c.copy(ip = x)
+ }
+ opt[String]("env") action { (x, c) =>
+ c.copy(env = Some(x))
+ } text("Comma-separated list of environmental variables (in 'FOO=BAR' " +
+ "format) to pass to the Spark execution environment.")
+ opt[Int]("port") action { (x, c) =>
+ c.copy(port = x)
+ } text("Port to bind to (default: 8000).")
+ opt[String]("engineInstanceId") required() action { (x, c) =>
+ c.copy(engineInstanceId = x)
+ } text("Engine instance ID.")
+ opt[Unit]("feedback") action { (_, c) =>
+ c.copy(feedback = true)
+ } text("Enable feedback loop to event server.")
+ opt[String]("event-server-ip") action { (x, c) =>
+ c.copy(eventServerIp = x)
+ }
+ opt[Int]("event-server-port") action { (x, c) =>
+ c.copy(eventServerPort = x)
+ } text("Event server port. Default: 7070")
+ opt[String]("accesskey") action { (x, c) =>
+ c.copy(accessKey = Some(x))
+ } text("Event server access key.")
+ opt[String]("log-url") action { (x, c) =>
+ c.copy(logUrl = Some(x))
+ }
+ opt[String]("log-prefix") action { (x, c) =>
+ c.copy(logPrefix = Some(x))
+ }
+ opt[String]("log-file") action { (x, c) =>
+ c.copy(logFile = Some(x))
+ }
+ opt[Unit]("verbose") action { (x, c) =>
+ c.copy(verbose = true)
+ } text("Enable verbose output.")
+ opt[Unit]("debug") action { (x, c) =>
+ c.copy(debug = true)
+ } text("Enable debug output.")
+ opt[String]("json-extractor") action { (x, c) =>
+ c.copy(jsonExtractor = JsonExtractorOption.withName(x))
+ }
+ }
+
+ parser.parse(args, ServerConfig()) map { sc =>
+ WorkflowUtils.modifyLogging(sc.verbose)
+ engineInstances.get(sc.engineInstanceId) map { engineInstance =>
+ val engineId = sc.engineId.getOrElse(engineInstance.engineId)
+ val engineVersion = sc.engineVersion.getOrElse(
+ engineInstance.engineVersion)
+ engineManifests.get(engineId, engineVersion) map { manifest =>
+ val engineFactoryName = engineInstance.engineFactory
- val upgrade = actorSystem.actorOf(Props(
- classOf[UpgradeActor],
- engineFactoryName))
- actorSystem.scheduler.schedule(
- 0.seconds,
- 1.days,
- upgrade,
- UpgradeCheck())
+ val master = actorSystem.actorOf(Props(
+ classOf[MasterActor],
+ sc,
+ engineInstance,
+ engineFactoryName,
+ manifest),
+ "master")
+ implicit val timeout = Timeout(5.seconds)
+ master ? StartServer()
+ actorSystem.awaitTermination
+ } getOrElse {
+ error(s"Invalid engine ID or version. Aborting server.")
+ }
+ } getOrElse {
+ error(s"Invalid engine instance ID. Aborting server.")
+ }
+ }
+ }
+
+ def createServerActorWithEngine[TD, EIN, PD, Q, P, A](
+ sc: ServerConfig,
+ engineInstance: EngineInstance,
+ engine: Engine[TD, EIN, PD, Q, P, A],
+ engineLanguage: EngineLanguage.Value,
+ manifest: EngineManifest): ActorRef = {
+
+ val engineParams = engine.engineInstanceToEngineParams(engineInstance, sc.jsonExtractor)
+
+ val kryo = KryoInstantiator.newKryoInjection
+
+ val modelsFromEngineInstance =
+ kryo.invert(modeldata.get(engineInstance.id).get.models).get.
+ asInstanceOf[Seq[Any]]
+
+ val batch = if (engineInstance.batch.nonEmpty) {
+ s"${engineInstance.engineFactory} (${engineInstance.batch})"
+ } else {
+ engineInstance.engineFactory
+ }
+
+ val sparkContext = WorkflowContext(
+ batch = batch,
+ executorEnv = engineInstance.env,
+ mode = "Serving",
+ sparkEnv = engineInstance.sparkConf)
+
+ val models = engine.prepareDeploy(
+ sparkContext,
+ engineParams,
+ engineInstance.id,
+ modelsFromEngineInstance,
+ params = WorkflowParams()
+ )
+
+ val algorithms = engineParams.algorithmParamsList.map { case (n, p) =>
+ Doer(engine.algorithmClassMap(n), p)
+ }
+
+ val servingParamsWithName = engineParams.servingParams
+
+ val serving = Doer(engine.servingClassMap(servingParamsWithName._1),
+ servingParamsWithName._2)
+
+ actorSystem.actorOf(
+ Props(
+ classOf[ServerActor[Q, P]],
+ sc,
+ engineInstance,
+ engine,
+ engineLanguage,
+ manifest,
+ engineParams.dataSourceParams._2,
+ engineParams.preparatorParams._2,
+ algorithms,
+ engineParams.algorithmParamsList.map(_._2),
+ models,
+ serving,
+ engineParams.servingParams._2))
+ }
+}
+
- class UpgradeActor(engineClass: String) extends Actor {
- val log = Logging(context.system, this)
- implicit val system = context.system
- def receive: Actor.Receive = {
- case x: UpgradeCheck =>
- WorkflowUtils.checkUpgrade("deployment", engineClass)
- }
- }
-
+class MasterActor (
+ sc: ServerConfig,
+ engineInstance: EngineInstance,
+ engineFactoryName: String,
+ manifest: EngineManifest) extends Actor with SSLConfiguration with KeyAuthentication {
+ val log = Logging(context.system, this)
+ implicit val system = context.system
+ var sprayHttpListener: Option[ActorRef] = None
+ var currentServerActor: Option[ActorRef] = None
+ var retry = 3
+
+ def undeploy(ip: String, port: Int): Unit = {
+ val serverUrl = s"https://${ip}:${port}"
+ log.info(
+ s"Undeploying any existing engine instance at $serverUrl")
+ try {
+ val code = scalaj.http.Http(s"$serverUrl/stop")
+ .option(HttpOptions.allowUnsafeSSL)
+ .param(ServerKey.param, ServerKey.get)
+ .method("POST").asString.code
+ code match {
+ case 200 => Unit
+ case 404 => log.error(
+ s"Another process is using $serverUrl. Unable to undeploy.")
+ case _ => log.error(
+ s"Another process is using $serverUrl, or an existing " +
+ s"engine server is not responding properly (HTTP $code). " +
+ "Unable to undeploy.")
+ }
+ } catch {
+ case e: java.net.ConnectException =>
+ log.warning(s"Nothing at $serverUrl")
+ case _: Throwable =>
+ log.error("Another process might be occupying " +
+ s"$ip:$port. Unable to undeploy.")
+ }
+ }
+
+ def receive: Actor.Receive = {
+ case x: StartServer =>
+ val actor = createServerActor(
+ sc,
+ engineInstance,
+ engineFactoryName,
+ manifest)
+ currentServerActor = Some(actor)
+ undeploy(sc.ip, sc.port)
+ self ! BindServer()
+ case x: BindServer =>
+ currentServerActor map { actor =>
+ val settings = ServerSettings(system)
+ IO(Http) ! Http.Bind(
+ actor,
+ interface = sc.ip,
+ port = sc.port,
+ settings = Some(settings.copy(sslEncryption = true)))
+ } getOrElse {
+ log.error("Cannot bind a non-existing server backend.")
+ }
+ case x: StopServer =>
+ log.info(s"Stop server command received.")
+ sprayHttpListener.map { l =>
+ log.info("Server is shutting down.")
+ l ! Http.Unbind(5.seconds)
+ system.shutdown
+ } getOrElse {
+ log.warning("No active server is running.")
+ }
+ case x: ReloadServer =>
+ log.info("Reload server command received.")
+ val latestEngineInstance =
+ CreateServer.engineInstances.getLatestCompleted(
+ manifest.id,
+ manifest.version,
+ engineInstance.engineVariant)
+ latestEngineInstance map { lr =>
+ val actor = createServerActor(sc, lr, engineFactoryName, manifest)
+ sprayHttpListener.map { l =>
+ l ! Http.Unbind(5.seconds)
+ val settings = ServerSettings(system)
+ IO(Http) ! Http.Bind(
+ actor,
+ interface = sc.ip,
+ port = sc.port,
+ settings = Some(settings.copy(sslEncryption = true)))
+ currentServerActor.get ! Kill
+ currentServerActor = Some(actor)
+ } getOrElse {
+ log.warning("No active server is running. Abort reloading.")
+ }
+ } getOrElse {
+ log.warning(
+ s"No latest completed engine instance for ${manifest.id} " +
+ s"${manifest.version}. Abort reloading.")
+ }
+ case x: Http.Bound =>
+ val serverUrl = s"https://${sc.ip}:${sc.port}"
+ log.info(s"Engine is deployed and running. Engine API is live at ${serverUrl}.")
+ sprayHttpListener = Some(sender)
+ case x: Http.CommandFailed =>
+ if (retry > 0) {
+ retry -= 1
+ log.error(s"Bind failed. Retrying... ($retry more trial(s))")
+ context.system.scheduler.scheduleOnce(1.seconds) {
+ self ! BindServer()
+ }
+ } else {
+ log.error("Bind failed. Shutting down.")
+ system.shutdown
+ }
+ }
+
+ def createServerActor(
+ sc: ServerConfig,
+ engineInstance: EngineInstance,
+ engineFactoryName: String,
+ manifest: EngineManifest): ActorRef = {
+ val (engineLanguage, engineFactory) =
+ WorkflowUtils.getEngine(engineFactoryName, getClass.getClassLoader)
+ val engine = engineFactory()
+
+ // EngineFactory return a base engine, which may not be deployable.
+ if (!engine.isInstanceOf[Engine[_,_,_,_,_,_]]) {
+ throw new NoSuchMethodException(s"Engine $engine is not deployable")
+ }
+
+ val deployableEngine = engine.asInstanceOf[Engine[_,_,_,_,_,_]]
+
+ CreateServer.createServerActorWithEngine(
+ sc,
+ engineInstance,
+ // engine,
+ deployableEngine,
+ engineLanguage,
+ manifest)
+ }
+}
+
+class ServerActor[Q, P](
+ val args: ServerConfig,
+ val engineInstance: EngineInstance,
+ val engine: Engine[_, _, _, Q, P, _],
+ val engineLanguage: EngineLanguage.Value,
+ val manifest: EngineManifest,
+ val dataSourceParams: Params,
+ val preparatorParams: Params,
+ val algorithms: Seq[BaseAlgorithm[_, _, Q, P]],
+ val algorithmsParams: Seq[Params],
+ val models: Seq[Any],
+ val serving: BaseServing[Q, P],
+ val servingParams: Params) extends Actor with HttpService with KeyAuthentication {
+ val serverStartTime = DateTime.now
+ val log = Logging(context.system, this)
+
+ var requestCount: Int = 0
+ var avgServingSec: Double = 0.0
+ var lastServingSec: Double = 0.0
+
+ /** The following is required by HttpService */
+ def actorRefFactory: ActorContext = context
+
+ implicit val timeout = Timeout(5, TimeUnit.SECONDS)
+ val pluginsActorRef =
+ context.actorOf(Props(classOf[PluginsActor], args.engineVariant), "PluginsActor")
+ val pluginContext = EngineServerPluginContext(log, args.engineVariant)
+
+ def receive: Actor.Receive = runRoute(myRoute)
+
+ val feedbackEnabled = if (args.feedback) {
+ if (args.accessKey.isEmpty) {
+ log.error("Feedback loop cannot be enabled because accessKey is empty.")
+ false
+ } else {
+ true
+ }
+ } else false
+
+ def remoteLog(logUrl: String, logPrefix: String, message: String): Unit = {
+ implicit val formats = Utils.json4sDefaultFormats
+ try {
+ scalaj.http.Http(logUrl).postData(
+ logPrefix + write(Map(
+ "engineInstance" -> engineInstance,
+ "message" -> message))).asString
+ } catch {
+ case e: Throwable =>
+ log.error(s"Unable to send remote log: ${e.getMessage}")
+ }
+ }
+
+ def getStackTraceString(e: Throwable): String = {
+ val writer = new StringWriter()
+ val printWriter = new PrintWriter(writer)
+ e.printStackTrace(printWriter)
+ writer.toString
+ }
+
+ val myRoute =
+ path("") {
+ get {
+ respondWithMediaType(`text/html`) {
+ detach() {
+ complete {
+ html.index(
+ args,
+ manifest,
+ engineInstance,
+ algorithms.map(_.toString),
+ algorithmsParams.map(_.toString),
+ models.map(_.toString),
+ dataSourceParams.toString,
+ preparatorParams.toString,
+ servingParams.toString,
+ serverStartTime,
+ feedbackEnabled,
+ args.eventServerIp,
+ args.eventServerPort,
+ requestCount,
+ avgServingSec,
+ lastServingSec
+ ).toString
+ }
+ }
+ }
+ }
+ } ~
+ path("queries.json") {
+ post {
+ detach() {
+ entity(as[String]) { queryString =>
+ try {
+ val servingStartTime = DateTime.now
+ val jsonExtractorOption = args.jsonExtractor
+ val queryTime = DateTime.now
+ // Extract Query from Json
+ val query = JsonExtractor.extract(
+ jsonExtractorOption,
+ queryString,
+ algorithms.head.queryClass,
+ algorithms.head.querySerializer,
+ algorithms.head.gsonTypeAdapterFactories
+ )
+ val queryJValue = JsonExtractor.toJValue(
+ jsonExtractorOption,
+ query,
+ algorithms.head.querySerializer,
+ algorithms.head.gsonTypeAdapterFactories)
+ // Deploy logic. First call Serving.supplement, then Algo.predict,
+ // finally Serving.serve.
+ val supplementedQuery = serving.supplementBase(query)
+ // TODO: Parallelize the following.
+ val predictions = algorithms.zipWithIndex.map { case (a, ai) =>
+ a.predictBase(models(ai), supplementedQuery)
+ }
+ // Notice that it is by design to call Serving.serve with the
+ // *original* query.
+ val prediction = serving.serveBase(query, predictions)
+ val predictionJValue = JsonExtractor.toJValue(
+ jsonExtractorOption,
+ prediction,
+ algorithms.head.querySerializer,
+ algorithms.head.gsonTypeAdapterFactories)
+ /** Handle feedback to Event Server
+ * Send the following back to the Event Server
+ * - appId
+ * - engineInstanceId
+ * - query
+ * - prediction
+ * - prId
+ */
+ val result = if (feedbackEnabled) {
+ implicit val formats =
+ algorithms.headOption map { alg =>
+ alg.querySerializer
+ } getOrElse {
+ Utils.json4sDefaultFormats
+ }
+ // val genPrId = Random.alphanumeric.take(64).mkString
+ def genPrId: String = Random.alphanumeric.take(64).mkString
+ val newPrId = prediction match {
+ case id: WithPrId =>
+ val org = id.prId
+ if (org.isEmpty) genPrId else org
+ case _ => genPrId
+ }
+
+ // also save Query's prId as prId of this pio_pr predict events
+ val queryPrId =
+ query match {
+ case id: WithPrId =>
+ Map("prId" -> id.prId)
+ case _ =>
+ Map()
+ }
+ val data = Map(
+ // "appId" -> dataSourceParams.asInstanceOf[ParamsWithAppId].appId,
+ "event" -> "predict",
+ "eventTime" -> queryTime.toString(),
+ "entityType" -> "pio_pr", // prediction result
+ "entityId" -> newPrId,
+ "properties" -> Map(
+ "engineInstanceId" -> engineInstance.id,
+ "query" -> query,
+ "prediction" -> prediction)) ++ queryPrId
+ // At this point args.accessKey should be Some(String).
+ val accessKey = args.accessKey.getOrElse("")
+ val f: Future[Int] = future {
+ scalaj.http.Http(
+ s"http://${args.eventServerIp}:${args.eventServerPort}/" +
+ s"events.json?accessKey=$accessKey").postData(
+ write(data)).header(
+ "content-type", "application/json").asString.code
+ }
+ f onComplete {
+ case Success(code) => {
+ if (code != 201) {
+ log.error(s"Feedback event failed. Status code: $code."
+ + s"Data: ${write(data)}.")
+ }
+ }
+ case Failure(t) => {
+ log.error(s"Feedback event failed: ${t.getMessage}") }
+ }
+ // overwrite prId in predictedResult
+ // - if it is WithPrId,
+ // then overwrite with new prId
+ // - if it is not WithPrId, no prId injection
+ if (prediction.isInstanceOf[WithPrId]) {
+ predictionJValue merge parse(s"""{"prId" : "$newPrId"}""")
+ } else {
+ predictionJValue
+ }
+ } else predictionJValue
+
+ val pluginResult =
+ pluginContext.outputBlockers.values.foldLeft(result) { case (r, p) =>
+ p.process(engineInstance, queryJValue, r, pluginContext)
+ }
+
+ // Bookkeeping
+ val servingEndTime = DateTime.now
+ lastServingSec =
+ (servingEndTime.getMillis - servingStartTime.getMillis) / 1000.0
+ avgServingSec =
+ ((avgServingSec * requestCount) + lastServingSec) /
+ (requestCount + 1)
+ requestCount += 1
+
+ respondWithMediaType(`application/json`) {
+ complete(compact(render(pluginResult)))
+ }
+ } catch {
+ case e: MappingException =>
+ log.error(
+ s"Query '$queryString' is invalid. Reason: ${e.getMessage}")
+ args.logUrl map { url =>
+ remoteLog(
+ url,
+ args.logPrefix.getOrElse(""),
+ s"Query:\n$queryString\n\nStack Trace:\n" +
+ s"${getStackTraceString(e)}\n\n")
+ }
+ complete(StatusCodes.BadRequest, e.getMessage)
+ case e: Throwable =>
+ val msg = s"Query:\n$queryString\n\nStack Trace:\n" +
+ s"${getStackTraceString(e)}\n\n"
+ log.error(msg)
+ args.logUrl map { url =>
+ remoteLog(
+ url,
+ args.logPrefix.getOrElse(""),
+ msg)
+ }
+ complete(StatusCodes.InternalServerError, msg)
+ }
+ }
+ }
+ }
+ } ~
+ path("reload") {
+ authenticate(withAccessKeyFromFile) { request =>
+ post {
+ complete {
+ context.actorSelection("/user/master") ! ReloadServer()
+ "Reloading..."
+ }
+ }
+ }
+ } ~
+ path("stop") {
+ authenticate(withAccessKeyFromFile) { request =>
+ post {
+ complete {
+ context.system.scheduler.scheduleOnce(1.seconds) {
+ context.actorSelection("/user/master") ! StopServer()
+ }
+ "Shutting down..."
+ }
+ }
+ }
+ } ~
+ pathPrefix("assets") {
+ getFromResourceDirectory("assets")
+ } ~
+ path("plugins.json") {
+ import EngineServerJson4sSupport._
+ get {
+ respondWithMediaType(MediaTypes.`application/json`) {
+ complete {
+ Map("plugins" -> Map(
+ "outputblockers" -> pluginContext.outputBlockers.map { case (n, p) =>
+ n -> Map(
+ "name" -> p.pluginName,
+ "description" -> p.pluginDescription,
+ "class" -> p.getClass.getName,
+ "params" -> pluginContext.pluginParams(p.pluginName))
+ },
+ "outputsniffers" -> pluginContext.outputSniffers.map { case (n, p) =>
+ n -> Map(
+ "name" -> p.pluginName,
+ "description" -> p.pluginDescription,
+ "class" -> p.getClass.getName,
+ "params" -> pluginContext.pluginParams(p.pluginName))
+ }
+ ))
+ }
+ }
+ }
+ } ~
+ path("plugins" / Segments) { segments =>
+ import EngineServerJson4sSupport._
+ get {
+ respondWithMediaType(MediaTypes.`application/json`) {
+ complete {
+ val pluginArgs = segments.drop(2)
+ val pluginType = segments(0)
+ val pluginName = segments(1)
+ pluginType match {
+ case EngineServerPlugin.outputSniffer =>
+ pluginsActorRef ? PluginsActor.HandleREST(
+ pluginName = pluginName,
+ pluginArgs = pluginArgs) map {
+ _.asInstanceOf[String]
+ }
+ }
+ }
+ }
+ }
+ }
+}
+
+object EngineServerJson4sSupport extends Json4sSupport {
+ implicit def json4sFormats: Formats = DefaultFormats
+}
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/c508c791/core/src/main/scala/org/apache/predictionio/workflow/WorkflowUtils.scala
----------------------------------------------------------------------
diff --cc core/src/main/scala/org/apache/predictionio/workflow/WorkflowUtils.scala
index cd80fd9,0000000..0df8db5
mode 100644,000000..100644
--- a/core/src/main/scala/org/apache/predictionio/workflow/WorkflowUtils.scala
+++ b/core/src/main/scala/org/apache/predictionio/workflow/WorkflowUtils.scala
@@@ -1,419 -1,0 +1,387 @@@
+/** Copyright 2015 TappingStone, Inc.
+ *
+ * Licensed 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.
+ */
+
+package org.apache.predictionio.workflow
+
+import java.io.File
+import java.io.FileNotFoundException
+
+import org.apache.predictionio.controller.EmptyParams
+import org.apache.predictionio.controller.EngineFactory
+import org.apache.predictionio.controller.EngineParamsGenerator
+import org.apache.predictionio.controller.Evaluation
+import org.apache.predictionio.controller.Params
+import org.apache.predictionio.controller.PersistentModelLoader
+import org.apache.predictionio.controller.Utils
+import org.apache.predictionio.core.BuildInfo
+
+import com.google.gson.Gson
+import com.google.gson.JsonSyntaxException
+import grizzled.slf4j.Logging
+import org.apache.predictionio.workflow.JsonExtractorOption.JsonExtractorOption
+import org.apache.log4j.Level
+import org.apache.log4j.LogManager
+import org.apache.spark.SparkContext
+import org.apache.spark.api.java.JavaRDDLike
+import org.apache.spark.rdd.RDD
+import org.json4s.JsonAST.JValue
+import org.json4s.MappingException
+import org.json4s._
+import org.json4s.native.JsonMethods._
+
- import scala.io.Source
+import scala.language.existentials
+import scala.reflect.runtime.universe
+
+/** Collection of reusable workflow related utilities. */
+object WorkflowUtils extends Logging {
+ @transient private lazy val gson = new Gson
+
+ /** Obtains an Engine object in Scala, or instantiate an Engine in Java.
+ *
+ * @param engine Engine factory name.
+ * @param cl A Java ClassLoader to look for engine-related classes.
+ *
+ * @throws ClassNotFoundException
+ * Thrown when engine factory class does not exist.
+ * @throws NoSuchMethodException
+ * Thrown when engine factory's apply() method is not implemented.
+ */
+ def getEngine(engine: String, cl: ClassLoader): (EngineLanguage.Value, EngineFactory) = {
+ val runtimeMirror = universe.runtimeMirror(cl)
+ val engineModule = runtimeMirror.staticModule(engine)
+ val engineObject = runtimeMirror.reflectModule(engineModule)
+ try {
+ (
+ EngineLanguage.Scala,
+ engineObject.instance.asInstanceOf[EngineFactory]
+ )
+ } catch {
+ case e @ (_: NoSuchFieldException | _: ClassNotFoundException) => try {
+ (
+ EngineLanguage.Java,
+ Class.forName(engine).newInstance.asInstanceOf[EngineFactory]
+ )
+ }
+ }
+ }
+
+ def getEngineParamsGenerator(epg: String, cl: ClassLoader):
+ (EngineLanguage.Value, EngineParamsGenerator) = {
+ val runtimeMirror = universe.runtimeMirror(cl)
+ val epgModule = runtimeMirror.staticModule(epg)
+ val epgObject = runtimeMirror.reflectModule(epgModule)
+ try {
+ (
+ EngineLanguage.Scala,
+ epgObject.instance.asInstanceOf[EngineParamsGenerator]
+ )
+ } catch {
+ case e @ (_: NoSuchFieldException | _: ClassNotFoundException) => try {
+ (
+ EngineLanguage.Java,
+ Class.forName(epg).newInstance.asInstanceOf[EngineParamsGenerator]
+ )
+ }
+ }
+ }
+
+ def getEvaluation(evaluation: String, cl: ClassLoader): (EngineLanguage.Value, Evaluation) = {
+ val runtimeMirror = universe.runtimeMirror(cl)
+ val evaluationModule = runtimeMirror.staticModule(evaluation)
+ val evaluationObject = runtimeMirror.reflectModule(evaluationModule)
+ try {
+ (
+ EngineLanguage.Scala,
+ evaluationObject.instance.asInstanceOf[Evaluation]
+ )
+ } catch {
+ case e @ (_: NoSuchFieldException | _: ClassNotFoundException) => try {
+ (
+ EngineLanguage.Java,
+ Class.forName(evaluation).newInstance.asInstanceOf[Evaluation]
+ )
+ }
+ }
+ }
+
+ /** Converts a JSON document to an instance of Params.
+ *
+ * @param language Engine's programming language.
+ * @param json JSON document.
+ * @param clazz Class of the component that is going to receive the resulting
+ * Params instance as a constructor argument.
+ * @param jsonExtractor JSON extractor option.
+ * @param formats JSON4S serializers for deserialization.
+ *
+ * @throws MappingException Thrown when JSON4S fails to perform conversion.
+ * @throws JsonSyntaxException Thrown when GSON fails to perform conversion.
+ */
+ def extractParams(
+ language: EngineLanguage.Value = EngineLanguage.Scala,
+ json: String,
+ clazz: Class[_],
+ jsonExtractor: JsonExtractorOption,
+ formats: Formats = Utils.json4sDefaultFormats): Params = {
+ implicit val f = formats
+ val pClass = clazz.getConstructors.head.getParameterTypes
+ if (pClass.size == 0) {
+ if (json != "") {
+ warn(s"Non-empty parameters supplied to ${clazz.getName}, but its " +
+ "constructor does not accept any arguments. Stubbing with empty " +
+ "parameters.")
+ }
+ EmptyParams()
+ } else {
+ val apClass = pClass.head
+ try {
+ JsonExtractor.extract(jsonExtractor, json, apClass, f).asInstanceOf[Params]
+ } catch {
+ case e@(_: MappingException | _: JsonSyntaxException) =>
+ error(
+ s"Unable to extract parameters for ${apClass.getName} from " +
+ s"JSON string: $json. Aborting workflow.",
+ e)
+ throw e
+ }
+ }
+ }
+
+ def getParamsFromJsonByFieldAndClass(
+ variantJson: JValue,
+ field: String,
+ classMap: Map[String, Class[_]],
+ engineLanguage: EngineLanguage.Value,
+ jsonExtractor: JsonExtractorOption): (String, Params) = {
+ variantJson findField {
+ case JField(f, _) => f == field
+ case _ => false
+ } map { jv =>
+ implicit lazy val formats = Utils.json4sDefaultFormats + new NameParamsSerializer
+ val np: NameParams = try {
+ jv._2.extract[NameParams]
+ } catch {
+ case e: Exception =>
+ error(s"Unable to extract $field name and params $jv")
+ throw e
+ }
+ val extractedParams = np.params.map { p =>
+ try {
+ if (!classMap.contains(np.name)) {
+ error(s"Unable to find $field class with name '${np.name}'" +
+ " defined in Engine.")
+ sys.exit(1)
+ }
+ WorkflowUtils.extractParams(
+ engineLanguage,
+ compact(render(p)),
+ classMap(np.name),
+ jsonExtractor,
+ formats)
+ } catch {
+ case e: Exception =>
+ error(s"Unable to extract $field params $p")
+ throw e
+ }
+ }.getOrElse(EmptyParams())
+
+ (np.name, extractedParams)
+ } getOrElse("", EmptyParams())
+ }
+
+ /** Grab environmental variables that starts with 'PIO_'. */
+ def pioEnvVars: Map[String, String] =
+ sys.env.filter(kv => kv._1.startsWith("PIO_"))
+
+ /** Converts Java (non-Scala) objects to a JSON4S JValue.
+ *
+ * @param params The Java object to be converted.
+ */
+ def javaObjectToJValue(params: AnyRef): JValue = parse(gson.toJson(params))
+
- private[predictionio] def checkUpgrade(
- component: String = "core",
- engine: String = ""): Unit = {
- val runner = new Thread(new UpgradeCheckRunner(component, engine))
- runner.start()
- }
-
+ // Extract debug string by recursively traversing the data.
+ def debugString[D](data: D): String = {
+ val s: String = data match {
+ case rdd: RDD[_] => {
+ debugString(rdd.collect())
+ }
+ case javaRdd: JavaRDDLike[_, _] => {
+ debugString(javaRdd.collect())
+ }
+ case array: Array[_] => {
+ "[" + array.map(debugString).mkString(",") + "]"
+ }
+ case d: AnyRef => {
+ d.toString
+ }
+ case null => "null"
+ }
+ s
+ }
+
+ /** Detect third party software configuration files to be submitted as
+ * extras to Apache Spark. This makes sure all executors receive the same
+ * configuration.
+ */
+ def thirdPartyConfFiles: Seq[String] = {
+ val thirdPartyFiles = Map(
+ "PIO_CONF_DIR" -> "log4j.properties",
+ "ES_CONF_DIR" -> "elasticsearch.yml",
+ "HADOOP_CONF_DIR" -> "core-site.xml",
+ "HBASE_CONF_DIR" -> "hbase-site.xml")
+
+ thirdPartyFiles.keys.toSeq.map { k: String =>
+ sys.env.get(k) map { x =>
+ val p = Seq(x, thirdPartyFiles(k)).mkString(File.separator)
+ if (new File(p).exists) Seq(p) else Seq[String]()
+ } getOrElse Seq[String]()
+ }.flatten
+ }
+
+ def thirdPartyClasspaths: Seq[String] = {
+ val thirdPartyPaths = Seq(
+ "PIO_CONF_DIR",
+ "ES_CONF_DIR",
+ "POSTGRES_JDBC_DRIVER",
+ "MYSQL_JDBC_DRIVER",
+ "HADOOP_CONF_DIR",
+ "HBASE_CONF_DIR")
+ thirdPartyPaths.map(p =>
+ sys.env.get(p).map(Seq(_)).getOrElse(Seq[String]())
+ ).flatten
+ }
+
+ def modifyLogging(verbose: Boolean): Unit = {
+ val rootLoggerLevel = if (verbose) Level.TRACE else Level.INFO
+ val chattyLoggerLevel = if (verbose) Level.INFO else Level.WARN
+
+ LogManager.getRootLogger.setLevel(rootLoggerLevel)
+
+ LogManager.getLogger("org.elasticsearch").setLevel(chattyLoggerLevel)
+ LogManager.getLogger("org.apache.hadoop").setLevel(chattyLoggerLevel)
+ LogManager.getLogger("org.apache.spark").setLevel(chattyLoggerLevel)
+ LogManager.getLogger("org.eclipse.jetty").setLevel(chattyLoggerLevel)
+ LogManager.getLogger("akka").setLevel(chattyLoggerLevel)
+ }
+
+ def extractNameParams(jv: JValue): NameParams = {
+ implicit val formats = Utils.json4sDefaultFormats
+ val nameOpt = (jv \ "name").extract[Option[String]]
+ val paramsOpt = (jv \ "params").extract[Option[JValue]]
+
+ if (nameOpt.isEmpty && paramsOpt.isEmpty) {
+ error("Unable to find 'name' or 'params' fields in" +
+ s" ${compact(render(jv))}.\n" +
+ "Since 0.8.4, the 'params' field is required in engine.json" +
+ " in order to specify parameters for DataSource, Preparator or" +
+ " Serving.\n" +
+ "Please go to https://docs.prediction.io/resources/upgrade/" +
+ " for detailed instruction of how to change engine.json.")
+ sys.exit(1)
+ }
+
+ if (nameOpt.isEmpty) {
+ info(s"No 'name' is found. Default empty String will be used.")
+ }
+
+ if (paramsOpt.isEmpty) {
+ info(s"No 'params' is found. Default EmptyParams will be used.")
+ }
+
+ NameParams(
+ name = nameOpt.getOrElse(""),
+ params = paramsOpt
+ )
+ }
+
+ def extractSparkConf(root: JValue): List[(String, String)] = {
+ def flatten(jv: JValue): List[(List[String], String)] = {
+ jv match {
+ case JObject(fields) =>
+ for ((namePrefix, childJV) <- fields;
+ (name, value) <- flatten(childJV))
+ yield (namePrefix :: name) -> value
+ case JArray(_) => {
+ error("Arrays are not allowed in the sparkConf section of engine.js.")
+ sys.exit(1)
+ }
+ case JNothing => List()
+ case _ => List(List() -> jv.values.toString)
+ }
+ }
+
+ flatten(root \ "sparkConf").map(x =>
+ (x._1.reduce((a, b) => s"$a.$b"), x._2))
+ }
+}
+
+case class NameParams(name: String, params: Option[JValue])
+
+class NameParamsSerializer extends CustomSerializer[NameParams](format => ( {
+ case jv: JValue => WorkflowUtils.extractNameParams(jv)
+}, {
+ case x: NameParams =>
+ JObject(JField("name", JString(x.name)) ::
+ JField("params", x.params.getOrElse(JNothing)) :: Nil)
+}
+ ))
+
+/** Collection of reusable workflow related utilities that touch on Apache
+ * Spark. They are separated to avoid compilation problems with certain code.
+ */
+object SparkWorkflowUtils extends Logging {
+ def getPersistentModel[AP <: Params, M](
+ pmm: PersistentModelManifest,
+ runId: String,
+ params: AP,
+ sc: Option[SparkContext],
+ cl: ClassLoader): M = {
+ val runtimeMirror = universe.runtimeMirror(cl)
+ val pmmModule = runtimeMirror.staticModule(pmm.className)
+ val pmmObject = runtimeMirror.reflectModule(pmmModule)
+ try {
+ pmmObject.instance.asInstanceOf[PersistentModelLoader[AP, M]](
+ runId,
+ params,
+ sc)
+ } catch {
+ case e @ (_: NoSuchFieldException | _: ClassNotFoundException) => try {
+ val loadMethod = Class.forName(pmm.className).getMethod(
+ "load",
+ classOf[String],
+ classOf[Params],
+ classOf[SparkContext])
+ loadMethod.invoke(null, runId, params, sc.orNull).asInstanceOf[M]
+ } catch {
+ case e: ClassNotFoundException =>
+ error(s"Model class ${pmm.className} cannot be found.")
+ throw e
+ case e: NoSuchMethodException =>
+ error(
+ "The load(String, Params, SparkContext) method cannot be found.")
+ throw e
+ }
+ }
+ }
+}
+
- class UpgradeCheckRunner(
- val component: String,
- val engine: String) extends Runnable with Logging {
- val version = BuildInfo.version
- val versionsHost = "https://direct.prediction.io/"
-
- def run(): Unit = {
- val url = if (engine == "") {
- s"$versionsHost$version/$component.json"
- } else {
- s"$versionsHost$version/$component/$engine.json"
- }
- try {
- val upgradeData = Source.fromURL(url)
- } catch {
- case e: FileNotFoundException =>
- debug(s"Update metainfo not found. $url")
- case e: java.net.UnknownHostException =>
- debug(s"${e.getClass.getName}: {e.getMessage}")
- }
- // TODO: Implement upgrade logic
- }
- }
-
+class WorkflowInterruption() extends Exception
+
+case class StopAfterReadInterruption() extends WorkflowInterruption
+
+case class StopAfterPrepareInterruption() extends WorkflowInterruption
+
+object EngineLanguage extends Enumeration {
+ val Scala, Java = Value
+}