You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by yl...@apache.org on 2016/12/01 04:32:30 UTC
spark git commit: [SPARK-18476][SPARKR][ML] SparkR Logistic
Regression should should support output original label.
Repository: spark
Updated Branches:
refs/heads/master 0a811210f -> 2eb6764fb
[SPARK-18476][SPARKR][ML] SparkR Logistic Regression should should support output original label.
## What changes were proposed in this pull request?
Similar to SPARK-18401, as a classification algorithm, logistic regression should support output original label instead of supporting index label.
In this PR, original label output is supported and test cases are modified and added. Document is also modified.
## How was this patch tested?
Unit tests.
Author: wm624@hotmail.com <wm...@hotmail.com>
Closes #15910 from wangmiao1981/audit.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/2eb6764f
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/2eb6764f
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/2eb6764f
Branch: refs/heads/master
Commit: 2eb6764fbb23553fc17772d8a4a1cad55ff7ba6e
Parents: 0a81121
Author: wm624@hotmail.com <wm...@hotmail.com>
Authored: Wed Nov 30 20:32:17 2016 -0800
Committer: Yanbo Liang <yb...@gmail.com>
Committed: Wed Nov 30 20:32:17 2016 -0800
----------------------------------------------------------------------
R/pkg/R/mllib.R | 19 +++++-----
R/pkg/inst/tests/testthat/test_mllib.R | 26 +++++++++-----
.../scala/org/apache/spark/SparkContext.scala | 2 +-
.../spark/ml/r/LogisticRegressionWrapper.scala | 37 ++++++++++++++------
4 files changed, 54 insertions(+), 30 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/2eb6764f/R/pkg/R/mllib.R
----------------------------------------------------------------------
diff --git a/R/pkg/R/mllib.R b/R/pkg/R/mllib.R
index 02bc645..eed8293 100644
--- a/R/pkg/R/mllib.R
+++ b/R/pkg/R/mllib.R
@@ -712,7 +712,6 @@ setMethod("predict", signature(object = "KMeansModel"),
#' of L1 and L2. Default is 0.0 which is an L2 penalty.
#' @param maxIter maximum iteration number.
#' @param tol convergence tolerance of iterations.
-#' @param fitIntercept whether to fit an intercept term.
#' @param family the name of family which is a description of the label distribution to be used in the model.
#' Supported options:
#' \itemize{
@@ -747,11 +746,11 @@ setMethod("predict", signature(object = "KMeansModel"),
#' \dontrun{
#' sparkR.session()
#' # binary logistic regression
-#' label <- c(1.0, 1.0, 1.0, 0.0, 0.0)
-#' feature <- c(1.1419053, 0.9194079, -0.9498666, -1.1069903, 0.2809776)
-#' binary_data <- as.data.frame(cbind(label, feature))
+#' label <- c(0.0, 0.0, 0.0, 1.0, 1.0)
+#' features <- c(1.1419053, 0.9194079, -0.9498666, -1.1069903, 0.2809776)
+#' binary_data <- as.data.frame(cbind(label, features))
#' binary_df <- createDataFrame(binary_data)
-#' blr_model <- spark.logit(binary_df, label ~ feature, thresholds = 1.0)
+#' blr_model <- spark.logit(binary_df, label ~ features, thresholds = 1.0)
#' blr_predict <- collect(select(predict(blr_model, binary_df), "prediction"))
#'
#' # summary of binary logistic regression
@@ -783,7 +782,7 @@ setMethod("predict", signature(object = "KMeansModel"),
#' @note spark.logit since 2.1.0
setMethod("spark.logit", signature(data = "SparkDataFrame", formula = "formula"),
function(data, formula, regParam = 0.0, elasticNetParam = 0.0, maxIter = 100,
- tol = 1E-6, fitIntercept = TRUE, family = "auto", standardization = TRUE,
+ tol = 1E-6, family = "auto", standardization = TRUE,
thresholds = 0.5, weightCol = NULL, aggregationDepth = 2,
probabilityCol = "probability") {
formula <- paste(deparse(formula), collapse = "")
@@ -795,10 +794,10 @@ setMethod("spark.logit", signature(data = "SparkDataFrame", formula = "formula")
jobj <- callJStatic("org.apache.spark.ml.r.LogisticRegressionWrapper", "fit",
data@sdf, formula, as.numeric(regParam),
as.numeric(elasticNetParam), as.integer(maxIter),
- as.numeric(tol), as.logical(fitIntercept),
- as.character(family), as.logical(standardization),
- as.array(thresholds), as.character(weightCol),
- as.integer(aggregationDepth), as.character(probabilityCol))
+ as.numeric(tol), as.character(family),
+ as.logical(standardization), as.array(thresholds),
+ as.character(weightCol), as.integer(aggregationDepth),
+ as.character(probabilityCol))
new("LogisticRegressionModel", jobj = jobj)
})
http://git-wip-us.apache.org/repos/asf/spark/blob/2eb6764f/R/pkg/inst/tests/testthat/test_mllib.R
----------------------------------------------------------------------
diff --git a/R/pkg/inst/tests/testthat/test_mllib.R b/R/pkg/inst/tests/testthat/test_mllib.R
index 467e00c..0553e70 100644
--- a/R/pkg/inst/tests/testthat/test_mllib.R
+++ b/R/pkg/inst/tests/testthat/test_mllib.R
@@ -646,30 +646,30 @@ test_that("spark.isotonicRegression", {
test_that("spark.logit", {
# test binary logistic regression
- label <- c(1.0, 1.0, 1.0, 0.0, 0.0)
+ label <- c(0.0, 0.0, 0.0, 1.0, 1.0)
feature <- c(1.1419053, 0.9194079, -0.9498666, -1.1069903, 0.2809776)
binary_data <- as.data.frame(cbind(label, feature))
binary_df <- createDataFrame(binary_data)
blr_model <- spark.logit(binary_df, label ~ feature, thresholds = 1.0)
blr_predict <- collect(select(predict(blr_model, binary_df), "prediction"))
- expect_equal(blr_predict$prediction, c(0, 0, 0, 0, 0))
+ expect_equal(blr_predict$prediction, c("0.0", "0.0", "0.0", "0.0", "0.0"))
blr_model1 <- spark.logit(binary_df, label ~ feature, thresholds = 0.0)
blr_predict1 <- collect(select(predict(blr_model1, binary_df), "prediction"))
- expect_equal(blr_predict1$prediction, c(1, 1, 1, 1, 1))
+ expect_equal(blr_predict1$prediction, c("1.0", "1.0", "1.0", "1.0", "1.0"))
# test summary of binary logistic regression
blr_summary <- summary(blr_model)
blr_fmeasure <- collect(select(blr_summary$fMeasureByThreshold, "threshold", "F-Measure"))
- expect_equal(blr_fmeasure$threshold, c(0.8221347, 0.7884005, 0.6674709, 0.3785437, 0.3434487),
+ expect_equal(blr_fmeasure$threshold, c(0.6565513, 0.6214563, 0.3325291, 0.2115995, 0.1778653),
tolerance = 1e-4)
- expect_equal(blr_fmeasure$"F-Measure", c(0.5000000, 0.8000000, 0.6666667, 0.8571429, 0.7500000),
+ expect_equal(blr_fmeasure$"F-Measure", c(0.6666667, 0.5000000, 0.8000000, 0.6666667, 0.5714286),
tolerance = 1e-4)
blr_precision <- collect(select(blr_summary$precisionByThreshold, "threshold", "precision"))
- expect_equal(blr_precision$precision, c(1.0000000, 1.0000000, 0.6666667, 0.7500000, 0.6000000),
+ expect_equal(blr_precision$precision, c(1.0000000, 0.5000000, 0.6666667, 0.5000000, 0.4000000),
tolerance = 1e-4)
blr_recall <- collect(select(blr_summary$recallByThreshold, "threshold", "recall"))
- expect_equal(blr_recall$recall, c(0.3333333, 0.6666667, 0.6666667, 1.0000000, 1.0000000),
+ expect_equal(blr_recall$recall, c(0.5000000, 0.5000000, 1.0000000, 1.0000000, 1.0000000),
tolerance = 1e-4)
# test model save and read
@@ -683,6 +683,16 @@ test_that("spark.logit", {
expect_error(summary(blr_model2))
unlink(modelPath)
+ # test prediction label as text
+ training <- suppressWarnings(createDataFrame(iris))
+ binomial_training <- training[training$Species %in% c("versicolor", "virginica"), ]
+ binomial_model <- spark.logit(binomial_training, Species ~ Sepal_Length + Sepal_Width)
+ prediction <- predict(binomial_model, binomial_training)
+ expect_equal(typeof(take(select(prediction, "prediction"), 1)$prediction), "character")
+ expected <- c("virginica", "virginica", "virginica", "versicolor", "virginica",
+ "versicolor", "virginica", "versicolor", "virginica", "versicolor")
+ expect_equal(as.list(take(select(prediction, "prediction"), 10))[[1]], expected)
+
# test multinomial logistic regression
label <- c(0.0, 1.0, 2.0, 0.0, 0.0)
feature1 <- c(4.845940, 5.64480, 7.430381, 6.464263, 5.555667)
@@ -694,7 +704,7 @@ test_that("spark.logit", {
model <- spark.logit(df, label ~., family = "multinomial", thresholds = c(0, 1, 1))
predict1 <- collect(select(predict(model, df), "prediction"))
- expect_equal(predict1$prediction, c(0, 0, 0, 0, 0))
+ expect_equal(predict1$prediction, c("0.0", "0.0", "0.0", "0.0", "0.0"))
# Summary of multinomial logistic regression is not implemented yet
expect_error(summary(model))
})
http://git-wip-us.apache.org/repos/asf/spark/blob/2eb6764f/core/src/main/scala/org/apache/spark/SparkContext.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/SparkContext.scala b/core/src/main/scala/org/apache/spark/SparkContext.scala
index 1cb39a4..b8414b5 100644
--- a/core/src/main/scala/org/apache/spark/SparkContext.scala
+++ b/core/src/main/scala/org/apache/spark/SparkContext.scala
@@ -19,7 +19,7 @@ package org.apache.spark
import java.io._
import java.lang.reflect.Constructor
-import java.net.{MalformedURLException, URI}
+import java.net.{URI}
import java.util.{Arrays, Locale, Properties, ServiceLoader, UUID}
import java.util.concurrent.{ConcurrentHashMap, ConcurrentMap}
import java.util.concurrent.atomic.{AtomicBoolean, AtomicInteger, AtomicReference}
http://git-wip-us.apache.org/repos/asf/spark/blob/2eb6764f/mllib/src/main/scala/org/apache/spark/ml/r/LogisticRegressionWrapper.scala
----------------------------------------------------------------------
diff --git a/mllib/src/main/scala/org/apache/spark/ml/r/LogisticRegressionWrapper.scala b/mllib/src/main/scala/org/apache/spark/ml/r/LogisticRegressionWrapper.scala
index 9b352c9..9fe6202 100644
--- a/mllib/src/main/scala/org/apache/spark/ml/r/LogisticRegressionWrapper.scala
+++ b/mllib/src/main/scala/org/apache/spark/ml/r/LogisticRegressionWrapper.scala
@@ -23,9 +23,9 @@ import org.json4s.JsonDSL._
import org.json4s.jackson.JsonMethods._
import org.apache.spark.ml.{Pipeline, PipelineModel}
-import org.apache.spark.ml.attribute.AttributeGroup
import org.apache.spark.ml.classification.{BinaryLogisticRegressionSummary, LogisticRegression, LogisticRegressionModel}
-import org.apache.spark.ml.feature.RFormula
+import org.apache.spark.ml.feature.{IndexToString, RFormula}
+import org.apache.spark.ml.r.RWrapperUtils._
import org.apache.spark.ml.util._
import org.apache.spark.sql.{DataFrame, Dataset}
@@ -34,6 +34,8 @@ private[r] class LogisticRegressionWrapper private (
val features: Array[String],
val isLoaded: Boolean = false) extends MLWritable {
+ import LogisticRegressionWrapper._
+
private val logisticRegressionModel: LogisticRegressionModel =
pipeline.stages(1).asInstanceOf[LogisticRegressionModel]
@@ -57,7 +59,11 @@ private[r] class LogisticRegressionWrapper private (
lazy val recallByThreshold: DataFrame = blrSummary.recallByThreshold
def transform(dataset: Dataset[_]): DataFrame = {
- pipeline.transform(dataset).drop(logisticRegressionModel.getFeaturesCol)
+ pipeline.transform(dataset)
+ .drop(PREDICTED_LABEL_INDEX_COL)
+ .drop(logisticRegressionModel.getFeaturesCol)
+ .drop(logisticRegressionModel.getLabelCol)
+
}
override def write: MLWriter = new LogisticRegressionWrapper.LogisticRegressionWrapperWriter(this)
@@ -66,6 +72,9 @@ private[r] class LogisticRegressionWrapper private (
private[r] object LogisticRegressionWrapper
extends MLReadable[LogisticRegressionWrapper] {
+ val PREDICTED_LABEL_INDEX_COL = "pred_label_idx"
+ val PREDICTED_LABEL_COL = "prediction"
+
def fit( // scalastyle:ignore
data: DataFrame,
formula: String,
@@ -73,7 +82,6 @@ private[r] object LogisticRegressionWrapper
elasticNetParam: Double,
maxIter: Int,
tol: Double,
- fitIntercept: Boolean,
family: String,
standardization: Boolean,
thresholds: Array[Double],
@@ -84,14 +92,14 @@ private[r] object LogisticRegressionWrapper
val rFormula = new RFormula()
.setFormula(formula)
- RWrapperUtils.checkDataColumns(rFormula, data)
+ .setForceIndexLabel(true)
+ checkDataColumns(rFormula, data)
val rFormulaModel = rFormula.fit(data)
- // get feature names from output schema
- val schema = rFormulaModel.transform(data).schema
- val featureAttrs = AttributeGroup.fromStructField(schema(rFormulaModel.getFeaturesCol))
- .attributes.get
- val features = featureAttrs.map(_.name.get)
+ val fitIntercept = rFormula.hasIntercept
+
+ // get labels and feature names from output schema
+ val (features, labels) = getFeaturesAndLabels(rFormulaModel, data)
// assemble and fit the pipeline
val logisticRegression = new LogisticRegression()
@@ -105,7 +113,9 @@ private[r] object LogisticRegressionWrapper
.setWeightCol(weightCol)
.setAggregationDepth(aggregationDepth)
.setFeaturesCol(rFormula.getFeaturesCol)
+ .setLabelCol(rFormula.getLabelCol)
.setProbabilityCol(probability)
+ .setPredictionCol(PREDICTED_LABEL_INDEX_COL)
if (thresholds.length > 1) {
logisticRegression.setThresholds(thresholds)
@@ -113,8 +123,13 @@ private[r] object LogisticRegressionWrapper
logisticRegression.setThreshold(thresholds(0))
}
+ val idxToStr = new IndexToString()
+ .setInputCol(PREDICTED_LABEL_INDEX_COL)
+ .setOutputCol(PREDICTED_LABEL_COL)
+ .setLabels(labels)
+
val pipeline = new Pipeline()
- .setStages(Array(rFormulaModel, logisticRegression))
+ .setStages(Array(rFormulaModel, logisticRegression, idxToStr))
.fit(data)
new LogisticRegressionWrapper(pipeline, features)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org