Custom Transformers

Every transformer in MLeap can be considered a customtransformer. The only difference between the transformers and bundleintegration code you write and what we write is that ours gets includedin the release jars. We welcome transformer additions to the MLeap project,please make a PR.

There are plenty of examples in the MLeap source codefor how to write your own transformers and make them serializableto/from Spark and MLeap.

Let’s go through a simple example of writing a custom transformer thatmaps an input string to a double using a Map[String, Double]to store the data needed for transformation. We will call our customtransformer: StringMap. This is a transformer that is included inMLeap source code, and you can view it here: StringMapModel.scala.

Overview

A brief overview of the steps involved:

  1. Build our core model logic that can be shared between Spark and MLeap
  2. Build the MLeap transformer
  3. Build the Spark transformer
  4. Build bundle serialization for MLeap
  5. Build bundle serialization for Spark
  6. Configure the MLeap Bundle registries with the MLeap and Sparkcustom transformer

Core Model

The core model is the logic needed to transform the input data. It has no dependencieson Spark or MLeap. In the case of our StringMapModel, it is a class thatknows how to map one string to a double. Let’s see what this lookslike in Scala.

StringMapModel.scala

  1. case class StringMapModel(labels: Map[String, Double]) extends Model {
  2.   def apply(label: String): Double = labels(label)
  4.   override def inputSchema: StructType = StructType("input" -> ScalarType.String).get
  6.   override def outputSchema: StructType = StructType("output" -> ScalarType.Double).get
  7. }

The case class has a set of labels that it know how to map to a double.This is very similar to a StringIndexerModel except that the values ofthe strings are arbitrary and not in sequence.

MLeap Transformer

The MLeap transformer is the piece of code that knows how to executeyour core model against a leap frame. All MLeap transformers inheritfrom a base class: ml.combust.mleap.runtime.transformer.Transformer.For our example StringMap transformer, we can use a utility base classfor simple input/output transformers called:ml.combust.mleap.runtime.transformer.SimpleTransformer. This baseclass takes care of a small amount of boilerplate for any transformerthat has exactly one input and one output column.

Here is the Scala code for the MLeap transformer.

StringMap.scala

  1. import ml.combust.mleap.core.feature.StringMapModel
  2. import ml.combust.mleap.core.types.NodeShape
  3. import ml.combust.mleap.runtime.function.UserDefinedFunction
  4. import ml.combust.mleap.runtime.transformer.{SimpleTransformer, Transformer}
  6. case class StringMap(override val uid: String = Transformer.uniqueName("string_map"),
  7.                      override val shape: NodeShape,
  8.                      override val model: StringMapModel) extends SimpleTransformer {
  9.   override val exec: UserDefinedFunction = (label: String) => model(label)

Note the UserDefinedFunction exec. This is an MLeapUserDefinedFunction that gets created from a Scala function usingreflection. UserDefinedFunctions are the primary way that MLeap allowsus to transform LeapFrames. The NodeShape shape defines the inputColand outputCol for this transformer.

Spark Transformer

The Spark transformer knows how to execute the core model against aSpark DataFrame. All Spark transformers inherit fromorg.apache.spark.ml.Transformer. If you have ever written a customSpark transformer before, this process will be very familiar.

Here is what a custom Spark transformer looks like in Scala.

StringMap.scala

  1. import ml.combust.mleap.core.feature.StringMapModel
  2. import org.apache.spark.annotation.DeveloperApi
  3. import org.apache.spark.ml.Transformer
  4. import org.apache.spark.ml.param.ParamMap
  5. import org.apache.spark.ml.param.shared.{HasInputCol, HasOutputCol}
  6. import org.apache.spark.ml.util.Identifiable
  7. import org.apache.spark.sql.functions._
  8. import org.apache.spark.sql.{DataFrame, Dataset}
  9. import org.apache.spark.sql.types._
  11. class StringMap(override val uid: String,
  12.                 val model: StringMapModel) extends Transformer
  13.   with HasInputCol
  14.   with HasOutputCol {
  15.   def this(model: StringMapModel) = this(uid = Identifiable.randomUID("string_map"), model = model)
  17.   def setInputCol(value: String): this.type = set(inputCol, value)
  18.   def setOutputCol(value: String): this.type = set(outputCol, value)
  20.   @org.apache.spark.annotation.Since("2.0.0")
  21.   override def transform(dataset: Dataset[_]): DataFrame = {
  22.     val stringMapUdf = udf {
  23.       (label: String) => model(label)
  24.     }
  26.     dataset.withColumn($(outputCol), stringMapUdf(dataset($(inputCol))))
  27.   }
  29.   override def copy(extra: ParamMap): Transformer = copyValues(new StringMap(uid, model), extra)
  31.   @DeveloperApi
  32.   override def transformSchema(schema: StructType): StructType = {
  33.     require(schema($(inputCol)).dataType.isInstanceOf[StringType],
  34.       s"Input column must be of type StringType but got ${schema($(inputCol)).dataType}")
  35.     val inputFields = schema.fields
  36.     require(!inputFields.exists(_.name == $(outputCol)),
  37.       s"Output column ${$(outputCol)} already exists.")
  39.     StructType(schema.fields :+ StructField($(outputCol), DoubleType))
  40.   }
  41. }

MLeap Serialization

We need to define how to serialize/deserialize our model to/from anMLeap Bundle. In order to do this, we make an implementation ofml.combust.mleap.bundle.ops.MleapOp and ml.combust.bundle.op.OpModel for ourMLeap transformer and core model, respectively. These type classes areall we need to define bundle serialization.

Here is what the serialization code looks like for our MLeap transformerin Scala.

NOTE: The code below looks long, but most of it is auto-generated by the IDE.

StringMapOp.scala

  1. import ml.combust.bundle.BundleContext
  2. import ml.combust.bundle.dsl._
  3. import ml.combust.bundle.op.OpModel
  4. import ml.combust.mleap.bundle.ops.MleapOp
  5. import ml.combust.mleap.core.feature.StringMapModel
  6. import ml.combust.mleap.runtime.MleapContext
  7. import ml.combust.mleap.runtime.transformer.feature.StringMap
  9. class StringMapOp extends MleapOp[StringMap, StringMapModel] {
  10.   override val Model: OpModel[MleapContext, StringMapModel] = new OpModel[MleapContext, StringMapModel] {
  11.     // the class of the model is needed for when we go to serialize JVM objects
  12.     override val klazz: Class[StringMapModel] = classOf[StringMapModel]
  14.     // a unique name for our op: "string_map"
  15.     override def opName: String = Bundle.BuiltinOps.feature.string_map
  17.     override def store(model: Model, obj: StringMapModel)
  18.                       (implicit context: BundleContext[MleapContext]): Model = {
  19.       // unzip our label map so we can store the label and the value
  20.       // as two parallel arrays, we do this because MLeap Bundles do
  21.       // not support storing data as a map
  22.       val (labels, values) = obj.labels.toSeq.unzip
  24.       // add the labels and values to the Bundle model that
  25.       // will be serialized to our MLeap bundle
  26.       model.withValue("labels", Value.stringList(labels)).
  27.         withValue("values", Value.doubleList(values))
  28.     }
  30.     override def load(model: Model)
  31.                      (implicit context: BundleContext[MleapContext]): StringMapModel = {
  32.       // retrieve our list of labels
  33.       val labels = model.value("labels").getStringList
  35.       // retrieve our list of values
  36.       val values = model.value("values").getDoubleList
  38.       // reconstruct the model using the parallel labels and values
  39.       StringMapModel(labels.zip(values).toMap)
  40.     }
  41.   }
  43.   // the core model that is used by the transformer
  44.   override def model(node: StringMap): StringMapModel = node.model
  45. }

We will need to register StringMapOp with the MLeap bundle registry atruntime to let MLeap know about it. We go over the registry later inthis article.

Spark Serialization

We also need to define how to serialize/deserialize the custom Sparktransformer to/from MLeap. This is very similar to the process we tookfor the MLeap transformer above. We will again be implementing bothml.combust.bundle.op.OpNode and ml.combust.bundle.op.OpModel.

Here is what the serialization code looks like for StringMap in Scala.

StringMapOp.scala

  1. import ml.combust.bundle.BundleContext
  2. import ml.combust.bundle.dsl._
  3. import ml.combust.bundle.op.{OpModel, OpNode}
  4. import ml.combust.mleap.core.feature.StringMapModel
  5. import ml.combust.mleap.runtime.MleapContext
  6. import org.apache.spark.ml.bundle.SparkBundleContext
  7. import org.apache.spark.ml.mleap.feature.StringMap
  9. class StringMapOp extends OpNode[SparkBundleContext, StringMap, StringMapModel] {
  10.   override val Model: OpModel[SparkBundleContext, StringMapModel] = new OpModel[SparkBundleContext, StringMapModel] {
  11.     // the class of the model is needed for when we go to serialize JVM objects
  12.     override val klazz: Class[StringMapModel] = classOf[StringMapModel]
  14.     // a unique name for our op: "string_map"
  15.     // this should be the same as for the MLeap transformer serialization
  16.     override def opName: String = Bundle.BuiltinOps.feature.string_map
  18.     override def store(model: Model, obj: StringMapModel)
  19.                       (implicit context: BundleContext[SparkBundleContext]): Model = {
  20.       // unzip our label map so we can store the label and the value
  21.       // as two parallel arrays, we do this because MLeap Bundles do
  22.       // not support storing data as a map
  23.       val (labels, values) = obj.labels.toSeq.unzip
  25.       // add the labels and values to the Bundle model that
  26.       // will be serialized to our MLeap bundle
  27.       model.withValue("labels", Value.stringList(labels)).
  28.         withValue("values", Value.doubleList(values))
  29.     }
  31.     override def load(model: Model)
  32.                      (implicit context: BundleContext[SparkBundleContext]): StringMapModel = {
  33.       // retrieve our list of labels
  34.       val labels = model.value("labels").getStringList
  36.       // retrieve our list of values
  37.       val values = model.value("values").getDoubleList
  39.       // reconstruct the model using the parallel labels and values
  40.       StringMapModel(labels.zip(values).toMap)
  41.     }
  42.   }
  43.   override val klazz: Class[StringMap] = classOf[StringMap]
  45.   override def name(node: StringMap): String = node.uid
  47.   override def model(node: StringMap): StringMapModel = node.model
  49.   override def load(node: Node, model: StringMapModel)
  50.                    (implicit context: BundleContext[SparkBundleContext]): StringMap = {
  51.     new StringMap(uid = node.name, model = model).
  52.       setInputCol(node.shape.standardInput.name).
  53.       setOutputCol(node.shape.standardOutput.name)
  54.   }
  56.   override def shape(node: StringMap)(implicit context: BundleContext[SparkBundleContext]): NodeShape =
  57.     NodeShape().withStandardIO(node.getInputCol, node.getOutputCol)
  58. }

We will need to register this with the MLeap registry as well, so thatMLeap knows how to serialize this Spark transformer.

MLeap Bundle Registries

A registry contains all of the custom transformers and types for a givenexecution engine. In this case, we support the MLeap and Spark executionengines for the StringMap transformer, so we will have to configureboth the Spark and MLeap registry to know how to serialize/deserializetheir respective transformers.

MLeap uses Typesafe Config toconfigure registries. By default, MLeap ships with registries configuredfor the Spark runtime and the MLeap runtime. You can take a look at eachof them here:

  1. MLeap Registry
  2. Spark Registry

By default, the MLeap runtime uses the configuration at ml.combust.mleap.registry.default.Spark uses the configuration at ml.combust.mleap.spark.registry.default.

MLeap Registry

In order to add the custom transformer to the default MLeap registry, we will add areference.conf file to our own project that looks like this:

  1. // make a list of all your custom transformers
  2. // the list contains the fully-qualified class names of the
  3. // OpNode implementations for your transformers
  4. my.domain.mleap.ops = ["my.domain.mleap.ops.StringMapOp"]
  6. // include the custom transformers we have defined to the default MLeap registry
  7. ml.combust.mleap.registry.default.ops += "my.domain.mleap.ops"

Spark Registry

In order to add the custom transformer to the default Spark registry, we will add areference.conf file to our own project that looks like this:

  1. // make a list of all your custom transformers
  2. // the list contains the fully-qualified class names of the
  3. // OpNode implementations for your transformers
  4. my.domain.mleap.spark.ops = ["my.domain.spark.ops.StringMapOp"]
  6. // include the custom transformers ops we have defined to the default Spark registries
  7. ml.combust.mleap.spark.registry.v20.ops += my.domain.mleap.spark.ops
  8. ml.combust.mleap.spark.registry.v21.ops += my.domain.mleap.spark.ops
  9. ml.combust.mleap.spark.registry.v22.ops += my.domain.mleap.spark.ops