rename to OffsetInstance and add param check

Author: actuaryzhang

mllib/src/main/scala/org/apache/spark/ml/feature/Instance.scala

@@ -27,3 +27,25 @@ import org.apache.spark.ml.linalg.Vector
  * @param features The vector of features for this data point.
  */
 private[ml] case class Instance(label: Double, weight: Double, features: Vector)
+
+/**
+ * Case class that represents an instance of data point with
+ * label, weight, offset and features.
+ *
+ * @param label Label for this data point.
+ * @param weight The weight of this instance.
+ * @param offset The offset used for this data point.
+ * @param features The vector of features for this data point.
+ */
+private[ml] case class OffsetInstance(label: Double, weight: Double, offset: Double,
+                                   features: Vector) {
+
+  /** Constructs from an [[Instance]] object and offset */
+  def this(instance: Instance, offset: Double = 0.0) = {
+    this(instance.label, instance.weight, offset, instance.features)
+  }
+
+  /** Converts to an [[Instance]] object by leaving out the offset. */
+  private[ml] def toInstance: Instance = Instance(label, weight, features)
+
+}

mllib/src/main/scala/org/apache/spark/ml/optim/IterativelyReweightedLeastSquares.scala

@@ -18,9 +18,8 @@
 package org.apache.spark.ml.optim
 
 import org.apache.spark.internal.Logging
-import org.apache.spark.ml.feature.Instance
+import org.apache.spark.ml.feature.{Instance, OffsetInstance}
 import org.apache.spark.ml.linalg._
-import org.apache.spark.ml.regression.GLRInstance
 import org.apache.spark.rdd.RDD
 
 /**
@@ -58,13 +57,13 @@ private[ml] class IterativelyReweightedLeastSquaresModel(
  */
 private[ml] class IterativelyReweightedLeastSquares(
     val initialModel: WeightedLeastSquaresModel,
-    val reweightFunc: (GLRInstance, WeightedLeastSquaresModel) => (Double, Double),
+    val reweightFunc: (OffsetInstance, WeightedLeastSquaresModel) => (Double, Double),
     val fitIntercept: Boolean,
     val regParam: Double,
     val maxIter: Int,
     val tol: Double) extends Logging with Serializable {
 
-  def fit(instances: RDD[GLRInstance]): IterativelyReweightedLeastSquaresModel = {
+  def fit(instances: RDD[OffsetInstance]): IterativelyReweightedLeastSquaresModel = {
 
     var converged = false
     var iter = 0

mllib/src/main/scala/org/apache/spark/ml/regression/GeneralizedLinearRegression.scala

@@ -24,7 +24,7 @@ import org.apache.spark.SparkException
 import org.apache.spark.annotation.{Experimental, Since}
 import org.apache.spark.internal.Logging
 import org.apache.spark.ml.PredictorParams
-import org.apache.spark.ml.feature.Instance
+import org.apache.spark.ml.feature.{Instance, OffsetInstance}
 import org.apache.spark.ml.linalg.{BLAS, Vector}
 import org.apache.spark.ml.optim._
 import org.apache.spark.ml.param._
@@ -123,6 +123,9 @@ private[regression] trait GeneralizedLinearRegressionBase extends PredictorParam
         s"with ${$(family)} family does not support ${$(link)} link function.")
     }
     val newSchema = super.validateAndTransformSchema(schema, fitting, featuresDataType)
+    if (isSet(offsetCol) && $(offsetCol).nonEmpty) {
+      SchemaUtils.checkNumericType(schema, $(offsetCol))
+    }
     if (hasLinkPredictionCol) {
       SchemaUtils.appendColumn(newSchema, $(linkPredictionCol), DoubleType)
     } else {
@@ -286,16 +289,16 @@ class GeneralizedLinearRegression @Since("2.0.0") (@Since("2.0.0") override val
 
     val w = if (!isDefined(weightCol) || $(weightCol).isEmpty) lit(1.0) else col($(weightCol))
     val off = if (!isDefined(offsetCol) || $(offsetCol).isEmpty) lit(0.0) else col($(offsetCol))
-    val instances: RDD[GLRInstance] =
+    val instances: RDD[OffsetInstance] =
       dataset.select(col($(labelCol)), w, off, col($(featuresCol))).rdd.map {
         case Row(label: Double, weight: Double, offset: Double, features: Vector) =>
-          new GLRInstance(label, weight, offset, features)
+          OffsetInstance(label, weight, offset, features)
       }
 
     val model = if (familyObj == Gaussian && linkObj == Identity) {
       // TODO: Make standardizeFeatures and standardizeLabel configurable.
       val wlsInstances: RDD[Instance] = instances.map { instance =>
-        new Instance(instance.label - instance.offset, instance.weight, instance.features)
+        Instance(instance.label - instance.offset, instance.weight, instance.features)
       }
       val optimizer = new WeightedLeastSquares($(fitIntercept), $(regParam), elasticNetParam = 0.0,
         standardizeFeatures = true, standardizeLabel = true)
@@ -365,7 +368,7 @@ object GeneralizedLinearRegression extends DefaultParamsReadable[GeneralizedLine
      * Get the initial guess model for [[IterativelyReweightedLeastSquares]].
      */
     def initialize(
-        instances: RDD[GLRInstance],
+        instances: RDD[OffsetInstance],
         fitIntercept: Boolean,
         regParam: Double): WeightedLeastSquaresModel = {
       val newInstances = instances.map { instance =>
@@ -384,11 +387,11 @@ object GeneralizedLinearRegression extends DefaultParamsReadable[GeneralizedLine
      * The reweight function used to update offsets and weights
      * at each iteration of [[IterativelyReweightedLeastSquares]].
      */
-    val reweightFunc: (GLRInstance, WeightedLeastSquaresModel) => (Double, Double) = {
-      (instance: GLRInstance, model: WeightedLeastSquaresModel) => {
-        val eta = model.predict(instance.features) + instance.offset
-        val mu = fitted(eta)
-        val newLabel = eta - instance.offset + (instance.label - mu) * link.deriv(mu)
+    val reweightFunc: (OffsetInstance, WeightedLeastSquaresModel) => (Double, Double) = {
+      (instance: OffsetInstance, model: WeightedLeastSquaresModel) => {
+        val eta = model.predict(instance.features)
+        val mu = fitted(eta + instance.offset)
+        val newLabel = eta + (instance.label - mu) * link.deriv(mu)
         val newWeight = instance.weight / (math.pow(this.link.deriv(mu), 2.0) * family.variance(mu))
         (newLabel, newWeight)
       }
@@ -766,7 +769,7 @@ class GeneralizedLinearRegressionModel private[ml] (
       val eta = BLAS.dot(features, coefficients) + intercept
       familyAndLink.fitted(eta)
     } else {
-      throw new SparkException("Must supply offset value when offset is set.")
+      throw new SparkException("Must supply offset to predict when offset column is set.")
     }
   }
 
@@ -1201,25 +1204,3 @@ class GeneralizedLinearRegressionTrainingSummary private[regression] (
     }
   }
 }
-
-/**
- * Case class that represents an instance of data point with
- * label, weight, offset and features.
- *
- * @param label Label for this data point.
- * @param weight The weight of this instance.
- * @param offset The offset used for this data point.
- * @param features The vector of features for this data point.
- */
-private[ml] case class GLRInstance(label: Double, weight: Double, offset: Double,
-                                   features: Vector) {
-
-  /** Constructs from an [[Instance]] object and offset */
-  def this(instance: Instance, offset: Double = 0.0) = {
-    this(instance.label, instance.weight, offset, instance.features)
-  }
-
-  /** Converts to an [[Instance]] object by leaving out the offset. */
-  private[ml] def toInstance: Instance = Instance(label, weight, features)
-
-}

mllib/src/test/scala/org/apache/spark/ml/optim/IterativelyReweightedLeastSquaresSuite.scala

@@ -18,17 +18,16 @@
 package org.apache.spark.ml.optim
 
 import org.apache.spark.SparkFunSuite
-import org.apache.spark.ml.feature.Instance
+import org.apache.spark.ml.feature.{Instance, OffsetInstance}
 import org.apache.spark.ml.linalg.Vectors
-import org.apache.spark.ml.regression.GLRInstance
 import org.apache.spark.ml.util.TestingUtils._
 import org.apache.spark.mllib.util.MLlibTestSparkContext
 import org.apache.spark.rdd.RDD
 
 class IterativelyReweightedLeastSquaresSuite extends SparkFunSuite with MLlibTestSparkContext {
 
-  private var instances1: RDD[GLRInstance] = _
-  private var instances2: RDD[GLRInstance] = _
+  private var instances1: RDD[OffsetInstance] = _
+  private var instances2: RDD[OffsetInstance] = _
 
   override def beforeAll(): Unit = {
     super.beforeAll()
@@ -44,7 +43,7 @@ class IterativelyReweightedLeastSquaresSuite extends SparkFunSuite with MLlibTes
       Instance(0.0, 2.0, Vectors.dense(1.0, 2.0)),
       Instance(1.0, 3.0, Vectors.dense(2.0, 1.0)),
       Instance(0.0, 4.0, Vectors.dense(3.0, 3.0))
-    ), 2).map(new GLRInstance(_))
+    ), 2).map(new OffsetInstance(_))
     /*
        R code:
 
@@ -57,7 +56,7 @@ class IterativelyReweightedLeastSquaresSuite extends SparkFunSuite with MLlibTes
       Instance(8.0, 2.0, Vectors.dense(1.0, 7.0)),
       Instance(3.0, 3.0, Vectors.dense(2.0, 11.0)),
       Instance(9.0, 4.0, Vectors.dense(3.0, 13.0))
-    ), 2).map(new GLRInstance(_))
+    ), 2).map(new OffsetInstance(_))
   }
 
   test("IRLS against GLM with Binomial errors") {
@@ -170,7 +169,7 @@ class IterativelyReweightedLeastSquaresSuite extends SparkFunSuite with MLlibTes
 object IterativelyReweightedLeastSquaresSuite {
 
   def BinomialReweightFunc(
-      instance: GLRInstance,
+      instance: OffsetInstance,
       model: WeightedLeastSquaresModel): (Double, Double) = {
     val eta = model.predict(instance.features) + instance.offset
     val mu = 1.0 / (1.0 + math.exp(-1.0 * eta))
@@ -180,7 +179,7 @@ object IterativelyReweightedLeastSquaresSuite {
   }
 
   def PoissonReweightFunc(
-      instance: GLRInstance,
+      instance: OffsetInstance,
       model: WeightedLeastSquaresModel): (Double, Double) = {
     val eta = model.predict(instance.features) + instance.offset
     val mu = math.exp(eta)
@@ -190,7 +189,7 @@ object IterativelyReweightedLeastSquaresSuite {
   }
 
   def L1RegressionReweightFunc(
-      instance: GLRInstance,
+      instance: OffsetInstance,
       model: WeightedLeastSquaresModel): (Double, Double) = {
     val eta = model.predict(instance.features) + instance.offset
     val e = math.max(math.abs(eta - instance.label), 1e-7)

mllib/src/test/scala/org/apache/spark/ml/regression/GeneralizedLinearRegressionSuite.scala

@@ -21,7 +21,7 @@ import scala.util.Random
 
 import org.apache.spark.SparkFunSuite
 import org.apache.spark.ml.classification.LogisticRegressionSuite._
-import org.apache.spark.ml.feature.Instance
+import org.apache.spark.ml.feature.{Instance, OffsetInstance}
 import org.apache.spark.ml.feature.LabeledPoint
 import org.apache.spark.ml.linalg.{BLAS, DenseVector, Vector, Vectors}
 import org.apache.spark.ml.param.{ParamMap, ParamsSuite}
@@ -604,10 +604,10 @@ class GeneralizedLinearRegressionSuite
       [1] -0.27378146  0.31599396 -0.06204946
      */
     val dataset = Seq(
-      GLRInstance(1.0, 1.0, 2.0, Vectors.dense(0.0, 5.0)),
-      GLRInstance(2.0, 2.0, 0.5, Vectors.dense(1.0, 2.0)),
-      GLRInstance(1.0, 3.0, 1.0, Vectors.dense(2.0, 1.0)),
-      GLRInstance(2.0, 4.0, 0.0, Vectors.dense(3.0, 3.0))
+      OffsetInstance(1.0, 1.0, 2.0, Vectors.dense(0.0, 5.0)),
+      OffsetInstance(2.0, 2.0, 0.5, Vectors.dense(1.0, 2.0)),
+      OffsetInstance(1.0, 3.0, 1.0, Vectors.dense(2.0, 1.0)),
+      OffsetInstance(2.0, 4.0, 0.0, Vectors.dense(3.0, 3.0))
     ).toDF()
 
     val expected = Seq(