I've built a Spark (2.2.0) ML pipeline with the output being a CrossValidatorModel, and using the save method to write the pipeline. I'd like to serve this pre-trained model using the Play (2.6.0) framework and Scala (2.11.11), but am having some problems figuring out how to use Spark with Play, and/or what's the best way to load the model.
Regarding my Play setup, the relevant content of my file structure is fairly straightforward:
app/
controllers/
HomeController.scala
ModelScorer.scala
models/
Passenger.scala
Prediction.scala
conf/
routes
where Passenger and Prediction are Case Classes representing the input and output of my model, respectively. HomeController represents the logic that will ingest a POST request in JSON format, parse the content into a Seq[Passenger], and feed that to ModelScorer.predict(data), as shown below.
// HomeController.scala
package controllers
import javax.inject._
import models.{Passenger, Prediction}
import play.api.mvc._
import play.api.libs.json._
import play.api.libs.functional.syntax._
@Singleton
class HomeController @Inject()(cc: ControllerComponents) extends AbstractController(cc) {
implicit val passengerReads: Reads[Passenger] = (
... // Various mappings
)(Passenger.apply _)
implicit val predictionWrites: Writes[Prediction] = (
... // Various mappings
)(unlift(Prediction.unapply))
def myEndpoint() = Action { implicit request: Request[AnyContent] =>
val inputData: JsValue = request.body.asJson.get
val passengers: Seq[Passenger] = inputData.validate[Seq[Passenger]].get
val predictions: Seq[Prediction] = ModelScorer.predict(passengers)
val outputData: JsValue = Json.toJson(predictions)
Ok(outputData)
}
}
To score predictions, the ModelScorer object is intended to initialize the SparkSession, load the model using Guava Cache, and then run the logic using the predict method to return predictions to HomeController. From what I can tell, the problematic line is val ds: Dataset[Passenger] = passengers.toDS which tells me something is wrong with my Spark initialization, but I'm unsure how to proceed.
// ModelScorer.scala
package controllers
import com.google.common.cache.{CacheBuilder, CacheLoader}
import models.{Passenger, Prediction}
import org.apache.log4j.{Level, Logger}
import org.apache.spark.ml.tuning.CrossValidatorModel
import org.apache.spark.sql.{Dataset, SparkSession}
object ModelScorer {
Logger.getLogger("org").setLevel(Level.OFF)
Logger.getLogger("akka").setLevel(Level.OFF)
val spark = SparkSession.builder
.master("local[*]")
.appName("ml-server")
.getOrCreate()
import spark.implicits._
val modelCache = CacheBuilder.newBuilder()
.build(
new CacheLoader[String, CrossValidatorModel] {
def load(path: String): CrossValidatorModel = {
CrossValidatorModel.load(path)
}
}
)
val model: CrossValidatorModel = modelCache.get("trained-cv-pipeline")
def predict(passengers: Seq[Passenger]): Seq[Prediction] = {
val ds: Dataset[Passenger] = passengers.toDS
val predictions: Seq[Prediction] = model.transform(ds)
.select("name","probability","prediction")
.withColumnRenamed("prediction","survives")
.as[Prediction]
.collect
.toSeq
predictions
}
}
Requisite dependencies in my build.sbt are:
libraryDependencies ++= Seq(
guice
, "org.scalatestplus.play" %% "scalatestplus-play" % "3.1.0" % Test
, "org.apache.spark" %% "spark-core" % "2.2.0"
, "org.apache.spark" %% "spark-sql" % "2.2.0"
, "org.apache.spark" %% "spark-mllib" % "2.2.0"
, "org.apache.hadoop" % "hadoop-client" % "2.7.2"
)
dependencyOverrides ++= Set(
"com.fasterxml.jackson.core" % "jackson-databind" % "2.6.5"
, "com.google.guava" % "guava" % "19.0"
)
Stacktrace, after POST request to http://localhost:9000/myEndpoint with requisite JSON, is:
@752mgi3ib - Internal server error, for (POST) [/myEndpoint] ->
play.api.http.HttpErrorHandlerExceptions$$anon$1: Execution
exception[[ScalaReflectionException: class models.Passenger in
JavaMirror with
DependencyClassLoader{file:/Users/XXXX/.ivy2/cache/org.scala-
lang/scala-library/jars/scala-library-2.11.11.jar,
...
... // Many, many lines
...
... :/Library/Java/JavaVirtualMachines/jdk1.8.0_131.jdk/Contents/Home/jre/classes] not found.
at play.api.http.HttpErrorHandlerExceptions$.throwableToUsefulException(HttpErrorHandler.scala:255)
at play.api.http.DefaultHttpErrorHandler.onServerError(HttpErrorHandler.scala:180)
at play.core.server.AkkaHttpServer$$anonfun$13$$anonfun$apply$1.applyOrElse(AkkaHttpServer.scala:252)
at play.core.server.AkkaHttpServer$$anonfun$13$$anonfun$apply$1.applyOrElse(AkkaHttpServer.scala:251)
at scala.concurrent.Future$$anonfun$recoverWith$1.apply(Future.scala:346)
at scala.concurrent.Future$$anonfun$recoverWith$1.apply(Future.scala:345)
at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:36)
at play.api.libs.streams.Execution$trampoline$.execute(Execution.scala:70)
at scala.concurrent.impl.CallbackRunnable.executeWithValue(Promise.scala:44)
at scala.concurrent.impl.Promise$DefaultPromise.scala$concurrent$impl$Promise$
DefaultPromise$$dispatchOrAddCallback(Promise.scala:284) Caused by: scala.ScalaReflectionException: class models.Passenger in JavaMirror with DependencyClassLoader{file:
...
... // Many, many lines
...
... :/Library/Java/JavaVirtualMachines/jdk1.8.0_131.jdk/Contents/Home/jre/classes] not found.
at scala.reflect.internal.Mirrors$RootsBase.staticClass(Mirrors.scala:123)
at scala.reflect.internal.Mirrors$RootsBase.staticClass(Mirrors.scala:22)
at controllers.ModelScorer$$typecreator3$1.apply(ModelScorer.scala:34)
at scala.reflect.api.TypeTags$WeakTypeTagImpl.tpe$lzycompute(TypeTags.scala:232)
at scala.reflect.api.TypeTags$WeakTypeTagImpl.tpe(TypeTags.scala:232)
at org.apache.spark.sql.catalyst.encoders.ExpressionEncoder$.apply(ExpressionEncoder.scala:49)
at org.apache.spark.sql.Encoders$.product(Encoders.scala:275)
at org.apache.spark.sql.LowPrioritySQLImplicits$class.newProductEncoder(SQLImplicits.scala:233)
at org.apache.spark.sql.SQLImplicits.newProductEncoder(SQLImplicits.scala:33)
at controllers.ModelScorer$.predict(ModelScorer.scala:34)
I've best traced the issue to the creation of the Dataset in ModelScorer.predict(passengers), more specifically the val ds: Dataset[Passenger] = passengers.toDS line, though I can get that line to run in the REPL using the sbt console, which makes me think it's some issue with integrating Spark into Play. A bit at a loss how to proceed, any and all guidance appreciated!