address comments

Author: actuaryzhang

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

@@ -31,6 +31,7 @@ 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.
+ * This is mainly used in GeneralizedLinearRegression currently.
  *
  * @param label Label for this data point.
  * @param weight The weight of this instance.

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

@@ -18,7 +18,7 @@
 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.rdd.RDD
 

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

@@ -135,8 +135,8 @@ private[regression] trait GeneralizedLinearRegressionBase extends PredictorParam
   def getLinkPredictionCol: String = $(linkPredictionCol)
 
   /**
-   * Param for offset column name. If this is not set or empty, we treat all
-   * instance offsets as 0.0.
+   * Param for offset column name. If this is not set or empty, we treat all instance offsets
+   * as 0.0. The feature specified as offset has a constant coefficient of 1.0.
    * @group param
    */
   final val offsetCol: Param[String] = new Param[String](this, "offsetCol", "The offset " +
@@ -145,6 +145,14 @@ private[regression] trait GeneralizedLinearRegressionBase extends PredictorParam
   /** @group getParam */
   def getOffsetCol: String = $(offsetCol)
 
+  /** Checks whether weight column is set and nonempty. */
+  private[regression] def hasWeightCol: Boolean =
+    isSet(weightCol) && $(weightCol).nonEmpty
+
+  /** Checks whether offset column is set and nonempty. */
+  private[regression] def hasOffsetCol: Boolean =
+    isSet(offsetCol) && $(offsetCol).nonEmpty
+
   /** Checks whether we should output link prediction. */
   private[regression] def hasLinkPredictionCol: Boolean = {
     isDefined(linkPredictionCol) && $(linkPredictionCol).nonEmpty
@@ -179,9 +187,11 @@ private[regression] trait GeneralizedLinearRegressionBase extends PredictorParam
     }
 
     val newSchema = super.validateAndTransformSchema(schema, fitting, featuresDataType)
-    if (fitting) {
-      if (isSetOffsetCol(this)) SchemaUtils.checkNumericType(schema, $(offsetCol))
+
+    if (hasOffsetCol) {
+      SchemaUtils.checkNumericType(schema, $(offsetCol))
     }
+
     if (hasLinkPredictionCol) {
       SchemaUtils.appendColumn(newSchema, $(linkPredictionCol), DoubleType)
     } else {
@@ -318,7 +328,6 @@ class GeneralizedLinearRegression @Since("2.0.0") (@Since("2.0.0") override val
 
   /**
    * Sets the value of param [[offsetCol]].
-   * The feature specified as offset has a constant coefficient of 1.0.
    * If this is not set or empty, we treat all instance offsets as 0.0.
    * Default is not set, so all instances have offset 0.0.
    *
@@ -364,8 +373,8 @@ class GeneralizedLinearRegression @Since("2.0.0") (@Since("2.0.0") override val
       "GeneralizedLinearRegression was given data with 0 features, and with Param fitIntercept " +
         "set to false. To fit a model with 0 features, fitIntercept must be set to true." )
 
-    val w = if (!isSetWeightCol(this)) lit(1.0) else col($(weightCol))
-    val offset = if (!isSetOffsetCol(this)) lit(0.0) else col($(offsetCol)).cast(DoubleType)
+    val w = if (!hasWeightCol) lit(1.0) else col($(weightCol))
+    val offset = if (!hasOffsetCol) lit(0.0) else col($(offsetCol)).cast(DoubleType)
 
     val model = if (familyAndLink.family == Gaussian && familyAndLink.link == Identity) {
       // TODO: Make standardizeFeatures and standardizeLabel configurable.
@@ -437,14 +446,6 @@ object GeneralizedLinearRegression extends DefaultParamsReadable[GeneralizedLine
 
   private[regression] val epsilon: Double = 1E-16
 
-  /** Checks whether weight column is set and nonempty */
-  private[regression] def isSetWeightCol(params: GeneralizedLinearRegressionBase): Boolean =
-    params.isSet(params.weightCol) && params.getWeightCol.nonEmpty
-
-  /** Checks whether offset column is set and nonempty */
-  private[regression] def isSetOffsetCol(params: GeneralizedLinearRegressionBase): Boolean =
-    params.isSet(params.offsetCol) && params.getOffsetCol.nonEmpty
-
   /**
    * Wrapper of family and link combination used in the model.
    */
@@ -476,14 +477,14 @@ object GeneralizedLinearRegression extends DefaultParamsReadable[GeneralizedLine
     }
 
     /**
-     * The reweight function used to update offsets and weights
+     * The reweight function used to update working labels and weights
      * at each iteration of [[IterativelyReweightedLeastSquares]].
      */
     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 eta = model.predict(instance.features) + instance.offset
+        val mu = fitted(eta)
+        val newLabel = eta - instance.offset + (instance.label - mu) * link.deriv(mu)
         val newWeight = instance.weight / (math.pow(this.link.deriv(mu), 2.0) * family.variance(mu))
         (newLabel, newWeight)
       }
@@ -989,7 +990,7 @@ class GeneralizedLinearRegressionModel private[ml] (
   /**
    * Calculates the predicted value when offset is set.
    */
-  protected def predict(features: Vector, offset: Double): Double = {
+  def predict(features: Vector, offset: Double): Double = {
     val eta = predictLink(features, offset)
     familyAndLink.fitted(eta)
   }
@@ -1009,22 +1010,8 @@ class GeneralizedLinearRegressionModel private[ml] (
   override protected def transformImpl(dataset: Dataset[_]): DataFrame = {
     val predictUDF = udf { (features: Vector, offset: Double) => predict(features, offset) }
     val predictLinkUDF = udf { (features: Vector, offset: Double) => predictLink(features, offset) }
-    /*
-     Offset is only validated when it's specified in the model and available in prediction data set.
-     When offset is specified but missing in the prediction data set, we default it to zero.
-     */
-    val offset = {
-      if (!isSetOffsetCol(this)) {
-        lit(0.0)
-      } else {
-        if (dataset.schema.fieldNames.contains($(offsetCol))) {
-          SchemaUtils.checkNumericType(dataset.schema, $(offsetCol))
-          col($(offsetCol)).cast(DoubleType)
-        } else {
-          lit(0.0)
-        }
-      }
-    }
+
+    val offset = if (!hasOffsetCol) lit(0.0) else col($(offsetCol)).cast(DoubleType)
     var output = dataset
     if ($(predictionCol).nonEmpty) {
       output = output.withColumn($(predictionCol), predictUDF(col($(featuresCol)), offset))
@@ -1218,11 +1205,11 @@ class GeneralizedLinearRegressionSummary private[regression] (
   private def prediction: Column = col(predictionCol)
 
   private def weight: Column = {
-    if (!isSetWeightCol(model)) lit(1.0) else col(model.getWeightCol)
+    if (!model.hasWeightCol) lit(1.0) else col(model.getWeightCol)
   }
 
   private def offset: Column = {
-    if (!isSetOffsetCol(model)) lit(0.0) else col(model.getOffsetCol).cast(DoubleType)
+    if (!model.hasOffsetCol) lit(0.0) else col(model.getOffsetCol).cast(DoubleType)
   }
 
   private[regression] lazy val devianceResiduals: DataFrame = {
@@ -1285,8 +1272,8 @@ class GeneralizedLinearRegressionSummary private[regression] (
         Estimate intercept analytically when there is no offset, or when there is offset but
         the model is Gaussian family with identity link. Otherwise, fit an intercept only model.
        */
-      if (!isSetOffsetCol(model) ||
-        (isSetOffsetCol(model) && family == Gaussian && link == Identity)) {
+      if (!model.hasOffsetCol ||
+        (model.hasOffsetCol && family == Gaussian && link == Identity)) {
         val agg = predictions.agg(sum(weight.multiply(
           label.minus(offset))), sum(weight)).first()
         link.link(agg.getDouble(0) / agg.getDouble(1))

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

@@ -39,11 +39,11 @@ class IterativelyReweightedLeastSquaresSuite extends SparkFunSuite with MLlibTes
        w <- c(1, 2, 3, 4)
      */
     instances1 = sc.parallelize(Seq(
-      Instance(1.0, 1.0, Vectors.dense(0.0, 5.0).toSparse),
-      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 OffsetInstance(_))
+      OffsetInstance(1.0, 1.0, 0.0, Vectors.dense(0.0, 5.0).toSparse),
+      OffsetInstance(0.0, 2.0, 0.0, Vectors.dense(1.0, 2.0)),
+      OffsetInstance(1.0, 3.0, 0.0, Vectors.dense(2.0, 1.0)),
+      OffsetInstance(0.0, 4.0, 0.0, Vectors.dense(3.0, 3.0))
+    ), 2)
     /*
        R code:
 
@@ -52,11 +52,11 @@ class IterativelyReweightedLeastSquaresSuite extends SparkFunSuite with MLlibTes
        w <- c(1, 2, 3, 4)
      */
     instances2 = sc.parallelize(Seq(
-      Instance(2.0, 1.0, Vectors.dense(0.0, 5.0).toSparse),
-      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 OffsetInstance(_))
+      OffsetInstance(2.0, 1.0, 0.0, Vectors.dense(0.0, 5.0).toSparse),
+      OffsetInstance(8.0, 2.0, 0.0, Vectors.dense(1.0, 7.0)),
+      OffsetInstance(3.0, 3.0, 0.0, Vectors.dense(2.0, 11.0)),
+      OffsetInstance(9.0, 4.0, 0.0, Vectors.dense(3.0, 13.0))
+    ), 2)
   }
 
   test("IRLS against GLM with Binomial errors") {

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

@@ -798,7 +798,7 @@ class GeneralizedLinearRegressionSuite
     }
   }
 
-  test("generalized linear regression with offset") {
+  test("generalized linear regression with weight and offset") {
     /*
       R code:
       library(statmod)
@@ -881,30 +881,6 @@ class GeneralizedLinearRegressionSuite
     }
   }
 
-  test("generalized linear regression: predict with no offset") {
-    val trainData = Seq(
-      OffsetInstance(2.0, 1.0, 2.0, Vectors.dense(0.0, 5.0)),
-      OffsetInstance(8.0, 2.0, 3.0, Vectors.dense(1.0, 7.0)),
-      OffsetInstance(3.0, 3.0, 1.0, Vectors.dense(2.0, 11.0)),
-      OffsetInstance(9.0, 4.0, 4.0, Vectors.dense(3.0, 13.0))
-    ).toDF()
-    val testData = trainData.select("weight", "features")
-
-    val trainer = new GeneralizedLinearRegression()
-      .setFamily("poisson")
-      .setWeightCol("weight")
-      .setOffsetCol("offset")
-      .setLinkPredictionCol("linkPrediction")
-
-    val model = trainer.fit(trainData)
-    model.transform(testData).select("features", "linkPrediction")
-      .collect().foreach {
-      case Row(features: DenseVector, linkPrediction1: Double) =>
-        val linkPrediction2 = BLAS.dot(features, model.coefficients) + model.intercept
-        assert(linkPrediction1 ~= linkPrediction2 relTol 1E-5, "Link Prediction mismatch")
-    }
-  }
-
   test("glm summary: gaussian family with weight and offset") {
     /*
       R code:
@@ -1309,7 +1285,7 @@ class GeneralizedLinearRegressionSuite
       -0.16134949  0.20807694 -0.22544551  0.03258777
       residuals(model, type = "working")
                  1            2            3            4
-       0.135315831 -0.084390309  0.113219135 -0.008279688
+      0.135315831 -0.084390309  0.113219135 -0.008279688
       residuals(model, type = "response")
                1          2          3          4
       -0.1923918  0.2565224 -0.1496381  0.0320653