[SPARK-20602] [ML]Adding LBFGS optimizer and Squared_hinge loss for LinearSVC

mllib/src/main/scala/org/apache/spark/ml/classification/LinearSVC.scala

@@ -17,10 +17,13 @@
 
 package org.apache.spark.ml.classification
 
+import java.util.Locale
+
 import scala.collection.mutable
 
 import breeze.linalg.{DenseVector => BDV}
-import breeze.optimize.{CachedDiffFunction, DiffFunction, OWLQN => BreezeOWLQN}
+import breeze.optimize.{CachedDiffFunction, DiffFunction, LBFGS => BreezeLBFGS,
+  OWLQN => BreezeOWLQN}
 import org.apache.hadoop.fs.Path
 
 import org.apache.spark.SparkException
@@ -42,7 +45,21 @@ import org.apache.spark.sql.functions.{col, lit}
 /** Params for linear SVM Classifier. */
 private[classification] trait LinearSVCParams extends ClassifierParams with HasRegParam
   with HasMaxIter with HasFitIntercept with HasTol with HasStandardization with HasWeightCol
-  with HasThreshold with HasAggregationDepth
+  with HasThreshold with HasAggregationDepth {
+
+  /**
+   * The optimization algorithm for LinearSVC.
+   * Supported options: "lbfgs" and "owlqn".
+   * (default: "lbfgs")
+   * @group param
+   */
+  final val optimizer: Param[String] = new Param[String](this, "optimizer", "The optimization" +
+    " algorithm to be used", ParamValidators.inArray[String](LinearSVC.supportedOptimizers))
+
+  /** @group getParam */
+  final def getOptimizer: String = $(optimizer)
+
+}
 
 /**
  * :: Experimental ::
@@ -60,6 +77,8 @@ class LinearSVC @Since("2.2.0") (
   extends Classifier[Vector, LinearSVC, LinearSVCModel]
   with LinearSVCParams with DefaultParamsWritable {
 
+  import LinearSVC._
+
   @Since("2.2.0")
   def this() = this(Identifiable.randomUID("linearsvc"))
 
@@ -145,6 +164,15 @@ class LinearSVC @Since("2.2.0") (
   def setAggregationDepth(value: Int): this.type = set(aggregationDepth, value)
   setDefault(aggregationDepth -> 2)
 
+  /**
+   * Set optimizer for LinearSVC. Supported options: "lbfgs" and "owlqn".
+   *
+   * @group setParam
+   */
+  @Since("2.2.0")
+  def setOptimizer(value: String): this.type = set(optimizer, value.toLowerCase(Locale.ROOT))
+  setDefault(optimizer -> "lbfgs")
+
   @Since("2.2.0")
   override def copy(extra: ParamMap): LinearSVC = defaultCopy(extra)
 
@@ -205,15 +233,21 @@ class LinearSVC @Since("2.2.0") (
       val costFun = new LinearSVCCostFun(instances, $(fitIntercept),
         $(standardization), bcFeaturesStd, regParamL2, $(aggregationDepth))
 
-      def regParamL1Fun = (index: Int) => 0D
-      val optimizer = new BreezeOWLQN[Int, BDV[Double]]($(maxIter), 10, regParamL1Fun, $(tol))
+      val optimizerAlgo = $(optimizer) match {
+        case LBFGS => new BreezeLBFGS[BDV[Double]]($(maxIter), 10, $(tol))
+        case OWLQN =>
+          def regParamL1Fun = (index: Int) => 0D
+          new BreezeOWLQN[Int, BDV[Double]]($(maxIter), 10, regParamL1Fun, $(tol))
+        case _ => throw new SparkException ("unexpected optimizer: " + $(optimizer))
+      }
+
       val initialCoefWithIntercept = Vectors.zeros(numFeaturesPlusIntercept)
 
-      val states = optimizer.iterations(new CachedDiffFunction(costFun),
+      val states = optimizerAlgo.iterations(new CachedDiffFunction(costFun),
         initialCoefWithIntercept.asBreeze.toDenseVector)
 
       val scaledObjectiveHistory = mutable.ArrayBuilder.make[Double]
-      var state: optimizer.State = null
+      var state: optimizerAlgo.State = null
       while (states.hasNext) {
         state = states.next()
         scaledObjectiveHistory += state.adjustedValue
@@ -258,6 +292,15 @@ class LinearSVC @Since("2.2.0") (
 @Since("2.2.0")
 object LinearSVC extends DefaultParamsReadable[LinearSVC] {
 
+  /** String name for Limited-memory BFGS. */
+  private[classification] val LBFGS: String = "lbfgs".toLowerCase(Locale.ROOT)
+
+  /** String name for Orthant-Wise Limited-memory Quasi-Newton. */
+  private[classification] val OWLQN: String = "owlqn".toLowerCase(Locale.ROOT)
+
+  /* Set of optimizers that LinearSVC supports */
+  private[classification] val supportedOptimizers = Array(LBFGS, OWLQN)
+
   @Since("2.2.0")
   override def load(path: String): LinearSVC = super.load(path)
 }

mllib/src/test/scala/org/apache/spark/ml/classification/LinearSVCSuite.scala

@@ -75,21 +75,25 @@ class LinearSVCSuite extends SparkFunSuite with MLlibTestSparkContext with Defau
   }
 
   test("Linear SVC binary classification") {
-    val svm = new LinearSVC()
-    val model = svm.fit(smallBinaryDataset)
-    assert(model.transform(smallValidationDataset)
-      .where("prediction=label").count() > nPoints * 0.8)
-    val sparseModel = svm.fit(smallSparseBinaryDataset)
-    checkModels(model, sparseModel)
+    LinearSVC.supportedOptimizers.foreach { opt =>
+      val svm = new LinearSVC().setOptimizer(opt)
+      val model = svm.fit(smallBinaryDataset)
+      assert(model.transform(smallValidationDataset)
+        .where("prediction=label").count() > nPoints * 0.8)
+      val sparseModel = svm.fit(smallSparseBinaryDataset)
+      checkModels(model, sparseModel)
+    }
   }
 
   test("Linear SVC binary classification with regularization") {
-    val svm = new LinearSVC()
-    val model = svm.setRegParam(0.1).fit(smallBinaryDataset)
-    assert(model.transform(smallValidationDataset)
-      .where("prediction=label").count() > nPoints * 0.8)
-    val sparseModel = svm.fit(smallSparseBinaryDataset)
-    checkModels(model, sparseModel)
+    LinearSVC.supportedOptimizers.foreach { opt =>
+      val svm = new LinearSVC().setOptimizer(opt).setMaxIter(10)
+      val model = svm.setRegParam(0.1).fit(smallBinaryDataset)
+      assert(model.transform(smallValidationDataset)
+        .where("prediction=label").count() > nPoints * 0.8)
+      val sparseModel = svm.fit(smallSparseBinaryDataset)
+      checkModels(model, sparseModel)
+    }
   }
 
   test("params") {
@@ -112,6 +116,7 @@ class LinearSVCSuite extends SparkFunSuite with MLlibTestSparkContext with Defau
     assert(lsvc.getFeaturesCol === "features")
     assert(lsvc.getPredictionCol === "prediction")
     assert(lsvc.getRawPredictionCol === "rawPrediction")
+    assert(lsvc.getOptimizer === "lbfgs")
     val model = lsvc.setMaxIter(5).fit(smallBinaryDataset)
     model.transform(smallBinaryDataset)
       .select("label", "prediction", "rawPrediction")
@@ -154,22 +159,23 @@ class LinearSVCSuite extends SparkFunSuite with MLlibTestSparkContext with Defau
 
   test("linearSVC with sample weights") {
     def modelEquals(m1: LinearSVCModel, m2: LinearSVCModel): Unit = {
-      assert(m1.coefficients ~== m2.coefficients absTol 0.05)
+      assert(m1.coefficients ~== m2.coefficients absTol 0.07)
       assert(m1.intercept ~== m2.intercept absTol 0.05)
     }
-
-    val estimator = new LinearSVC().setRegParam(0.01).setTol(0.01)
-    val dataset = smallBinaryDataset
-    MLTestingUtils.testArbitrarilyScaledWeights[LinearSVCModel, LinearSVC](
-      dataset.as[LabeledPoint], estimator, modelEquals)
-    MLTestingUtils.testOutliersWithSmallWeights[LinearSVCModel, LinearSVC](
-      dataset.as[LabeledPoint], estimator, 2, modelEquals, outlierRatio = 3)
-    MLTestingUtils.testOversamplingVsWeighting[LinearSVCModel, LinearSVC](
-      dataset.as[LabeledPoint], estimator, modelEquals, 42L)
+    LinearSVC.supportedOptimizers.foreach { opt =>
+      val estimator = new LinearSVC().setRegParam(0.02).setTol(0.01).setOptimizer(opt)
+      val dataset = smallBinaryDataset
+      MLTestingUtils.testArbitrarilyScaledWeights[LinearSVCModel, LinearSVC](
+        dataset.as[LabeledPoint], estimator, modelEquals)
+      MLTestingUtils.testOutliersWithSmallWeights[LinearSVCModel, LinearSVC](
+        dataset.as[LabeledPoint], estimator, 2, modelEquals, outlierRatio = 3)
+      MLTestingUtils.testOversamplingVsWeighting[LinearSVCModel, LinearSVC](
+        dataset.as[LabeledPoint], estimator, modelEquals, 42L)
+    }
   }
 
-  test("linearSVC comparison with R e1071 and scikit-learn") {
-    val trainer1 = new LinearSVC()
+  test("linearSVC OWLQN comparison with R e1071 and scikit-learn") {
+    val trainer1 = new LinearSVC().setOptimizer("owlqn")
       .setRegParam(0.00002) // set regParam = 2.0 / datasize / c
       .setMaxIter(200)
       .setTol(1e-4)
@@ -223,6 +229,25 @@ class LinearSVCSuite extends SparkFunSuite with MLlibTestSparkContext with Defau
     assert(model1.coefficients ~== coefficientsSK relTol 4E-3)
   }
 
+  test("linearSVC LBFGS comparison with R e1071 and scikit-learn") {
+    val trainer1 = new LinearSVC().setOptimizer("LBFGS")
+      .setRegParam(0.00003)
+      .setMaxIter(200)
+      .setTol(1e-4)
+    val model1 = trainer1.fit(binaryDataset)
+
+    // refer to last unit test for R and python code
+    val coefficientsR = Vectors.dense(7.310338, 14.89741, 22.21005, 29.83508)
+    val interceptR = 7.440177
+    assert(model1.intercept ~== interceptR relTol 2E-2)
+    assert(model1.coefficients ~== coefficientsR relTol 1E-2)
+
+    val coefficientsSK = Vectors.dense(7.24690165, 14.77029087, 21.99924004, 29.5575729)
+    val interceptSK = 7.36947518
+    assert(model1.intercept ~== interceptSK relTol 1E-2)
+    assert(model1.coefficients ~== coefficientsSK relTol 1E-2)
+  }
+
   test("read/write: SVM") {
     def checkModelData(model: LinearSVCModel, model2: LinearSVCModel): Unit = {
       assert(model.intercept === model2.intercept)