Deserialization of Linear Regression Broken for Spark 2.3

[tcoatale]

It is no longer possible to use a linear regression model having been serialized and then deserialized using mleap in spark 2.3.

The following sequence of operations fails

    import ml.combust.bundle.BundleFile
    import ml.combust.mleap.spark.SparkSupport._
    import org.apache.spark.ml.Pipeline
    import org.apache.spark.ml.bundle.SparkBundleContext
    import resource._
    import spark.implicits._

    if (new File("/tmp/simple-spark-pipeline.zip").exists())
      new File("/tmp/simple-spark-pipeline.zip").delete()

    val trainingData = Seq((1, 0.8, 1.2), (1, 0.7, 1.0), (1, 1.2, 1.1), (1, 0.7, 2.0), (2, 1.2, 1.0))
      .toDF("feature1", "feature2", "value")

    val testingData = Seq((1, 0.8, 1.2), (1, 0.7, 1.0), (1, 1.2, 1.1), (1, 0.7, 2.0), (2, 1.2, 1.0))
      .toDF("feature1", "feature2", "value")

    val vectorAssembler = new VectorAssembler()
      .setInputCols(Array("feature1", "feature2"))
      .setOutputCol("features")

    val assembledData = vectorAssembler.transform(trainingData)

    val lr = new LinearRegression()
      .setFeaturesCol("features")
      .setLabelCol("value")

    val model = lr.fit(assembledData)

    val pipeline = new Pipeline()
      .setStages(Array(vectorAssembler, model))
      .fit(trainingData)

    // then serialize pipeline
    val sbc = SparkBundleContext().withDataset(pipeline.transform(trainingData))
    for(bf <- managed(BundleFile("jar:file:/tmp/simple-spark-pipeline.zip"))) {
      pipeline.writeBundle.save(bf)(sbc).get
    }

    import ml.combust.bundle.BundleFile
    import ml.combust.mleap.runtime.MleapSupport._
    import resource._

    val bundle = (for(bundleFile <- managed(BundleFile("jar:file:/tmp/simple-spark-pipeline.zip"))) yield {
      bundleFile.loadMleapBundle().get
    }).opt.get
    // transform the dataframe using our pipeline
    val mleapPipeline = bundle.root

    // then serialize pipeline
    for (bundle <- managed(BundleFile("jar:file:/tmp/simple-mleap-pipeline.zip"))) {
      mleapPipeline.writeBundle.format(SerializationFormat.Json).save(bundle)
    }

    val sparkBundle = (for(bf <- managed(BundleFile("jar:file:/tmp/simple-mleap-pipeline.zip"))) yield {
      bf.loadSparkBundle().get
    }).either

    val transformer = sparkBundle.right.get.root
    transformer.transform(testingData)

with the following stacktrace:

Failed to find a default value for loss
java.util.NoSuchElementException: Failed to find a default value for loss
	at org.apache.spark.ml.param.Params$$anonfun$getOrDefault$2.apply(params.scala:780)
	at org.apache.spark.ml.param.Params$$anonfun$getOrDefault$2.apply(params.scala:780)
	at scala.Option.getOrElse(Option.scala:121)
	at org.apache.spark.ml.param.Params$class.getOrDefault(params.scala:779)
	at org.apache.spark.ml.PipelineStage.getOrDefault(Pipeline.scala:42)
	at org.apache.spark.ml.param.Params$class.$(params.scala:786)
	at org.apache.spark.ml.PipelineStage.$(Pipeline.scala:42)
	at org.apache.spark.ml.regression.LinearRegressionParams$class.validateAndTransformSchema(LinearRegression.scala:110)
	at org.apache.spark.ml.regression.LinearRegressionModel.validateAndTransformSchema(LinearRegression.scala:640)
	at org.apache.spark.ml.PredictionModel.transformSchema(Predictor.scala:192)
	at org.apache.spark.ml.PipelineModel$$anonfun$transformSchema$5.apply(Pipeline.scala:311)
	at org.apache.spark.ml.PipelineModel$$anonfun$transformSchema$5.apply(Pipeline.scala:311)
	at scala.collection.IndexedSeqOptimized$class.foldl(IndexedSeqOptimized.scala:57)
	at scala.collection.IndexedSeqOptimized$class.foldLeft(IndexedSeqOptimized.scala:66)
	at scala.collection.mutable.ArrayOps$ofRef.foldLeft(ArrayOps.scala:186)
	at org.apache.spark.ml.PipelineModel.transformSchema(Pipeline.scala:311)
	at org.apache.spark.ml.PipelineStage.transformSchema(Pipeline.scala:74)
	at org.apache.spark.ml.PipelineModel.transform(Pipeline.scala:305)

In Spark 2.3.0, a new param was introduced: loss. When using a LinearRegressionModel's method transform, the method validateAndTransformSchema checks the compatibility between the solver param and the loss param, which, in our case, breaks because the deserialized model does not contain those parameters.

Quite frankly, it seems like this could be handled better by spark because the check concerns the training of the model and is only being done once the model is already trained. However, something should probably be done to make sure the export/import feature is functional for linear regression.

Please let me know what you think. I am open to contributing a PR to this issue, as long as you let me know how you would like to approach this issue.

[tcoatale]

Hi, any update on this issue?

[Excessionist]

I am stuck with the same problem. In my case, I have an old model which is almost impossible to rebuild, but which now no longer works. I am at a total loss of what to do to fix it.

[AndyMc629]

I've also had this issue when deserializing a model in the python binding.

import mleap.pyspark
from mleap.pyspark.spark_support import SimpleSparkSerializer

cvModel.bestModel.serializeToBundle("jar:file:/tmp/mleap_python_model_export/test_pipeline-json.zip", cvModel.bestModel.transform(train_data))

Then running

from pyspark.ml import PipelineModel
deserializedPipeline = PipelineModel.deserializeFromBundle("jar:file:/tmp/mleap_python_model_export/test_pipeline-json.zip")
deserializedPipeline.transform(test_data)

Gives the following stacktrace (in Databricks):

java.util.NoSuchElementException: Failed to find a default value for family

---------------------------------------------------------------------------
Py4JJavaError                             Traceback (most recent call last)
<command-3835588770737914> in <module>()
----> 1 deserializedPipeline.transform(test_data)

/databricks/spark/python/pyspark/ml/base.py in transform(self, dataset, params)
    171                 return self.copy(params)._transform(dataset)
    172             else:
--> 173                 return self._transform(dataset)
    174         else:
    175             raise ValueError("Params must be a param map but got %s." % type(params))

/databricks/spark/python/pyspark/ml/pipeline.py in _transform(self, dataset)
    260     def _transform(self, dataset):
    261         for t in self.stages:
--> 262             dataset = t.transform(dataset)
    263         return dataset
    264 

/databricks/spark/python/pyspark/ml/base.py in transform(self, dataset, params)
    171                 return self.copy(params)._transform(dataset)
    172             else:
--> 173                 return self._transform(dataset)
    174         else:
    175             raise ValueError("Params must be a param map but got %s." % type(params))

/databricks/spark/python/pyspark/ml/wrapper.py in _transform(self, dataset)
    303     def _transform(self, dataset):
    304         self._transfer_params_to_java()
--> 305         return DataFrame(self._java_obj.transform(dataset._jdf), dataset.sql_ctx)
    306 
    307 

/databricks/spark/python/lib/py4j-0.10.6-src.zip/py4j/java_gateway.py in __call__(self, *args)
   1158         answer = self.gateway_client.send_command(command)
   1159         return_value = get_return_value(
-> 1160             answer, self.gateway_client, self.target_id, self.name)
   1161 
   1162         for temp_arg in temp_args:

/databricks/spark/python/pyspark/sql/utils.py in deco(*a, **kw)
     61     def deco(*a, **kw):
     62         try:
---> 63             return f(*a, **kw)
     64         except py4j.protocol.Py4JJavaError as e:
     65             s = e.java_exception.toString()

/databricks/spark/python/lib/py4j-0.10.6-src.zip/py4j/protocol.py in get_return_value(answer, gateway_client, target_id, name)
    318                 raise Py4JJavaError(
    319                     "An error occurred while calling {0}{1}{2}.\n".
--> 320                     format(target_id, ".", name), value)
    321             else:
    322                 raise Py4JError(

Py4JJavaError: An error occurred while calling o6515.transform.
: java.util.NoSuchElementException: Failed to find a default value for family
	at org.apache.spark.ml.param.Params$$anonfun$getOrDefault$2.apply(params.scala:780)
	at org.apache.spark.ml.param.Params$$anonfun$getOrDefault$2.apply(params.scala:780)
	at scala.Option.getOrElse(Option.scala:121)
	at org.apache.spark.ml.param.Params$class.getOrDefault(params.scala:779)
	at org.apache.spark.ml.PipelineStage.getOrDefault(Pipeline.scala:42)
	at org.apache.spark.ml.param.Params$class.$(params.scala:786)
	at org.apache.spark.ml.PipelineStage.$(Pipeline.scala:42)
	at org.apache.spark.ml.regression.GeneralizedLinearRegressionBase$class.validateAndTransformSchema(GeneralizedLinearRegression.scala:187)
	at org.apache.spark.ml.regression.GeneralizedLinearRegressionModel.validateAndTransformSchema(GeneralizedLinearRegression.scala:994)
	at org.apache.spark.ml.PredictionModel.transformSchema(Predictor.scala:192)
	at org.apache.spark.ml.regression.GeneralizedLinearRegressionModel.transform(GeneralizedLinearRegression.scala:1033)
	at sun.reflect.GeneratedMethodAccessor675.invoke(Unknown Source)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:498)
	at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
	at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:380)
	at py4j.Gateway.invoke(Gateway.java:295)
	at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
	at py4j.commands.CallCommand.execute(CallCommand.java:79)
	at py4j.GatewayConnection.run(GatewayConnection.java:226)
	at java.lang.Thread.run(Thread.java:748)

A solution or even a temp workaround for this would be great.

[tcoatale]

@AndyMc629 , a possible but far from ideal workaround would be to systematically store your model in spark format (through the write method of PipelineModel) and mleap format. You would then load the spark format if you want to use it in spark or in mleap format if you want to use it in mleap at runtime.

It's not ideal and there are redundancies, but until that's fixed, I don't really see any alternative.
Hope that helps

[ancasarb]

Hey @tcoatale I've raised an issue with Spark (https://issues.apache.org/jira/browse/SPARK-27621) and created a PR for the fix, you can follow progress there. I think the right thing to do here is to fix it within Spark, as mleap really shouldn't serialize any training related params.

In the meantime, I think the solution to also store the model in Spark format for scoring within Spark seems like the available workaround at the moment.

[ancasarb]

The Spark PR apache/spark#24509 got merged into master/2.3/24.