Skip to content

[SPARK-13677][ML] Implement Tree-Based Feature Transformation for ML #25383

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
zhengruifeng wants to merge 19 commits into from
Closed

[SPARK-13677][ML] Implement Tree-Based Feature Transformation for ML #25383

Show file tree
Hide file tree
Changes from 17 commits
Commits
Show all changes
19 commits
Select commit Hold shift + click to select a range
95b1d22
init
zhengruifeng Jun 5, 2019
c17af00
xxx
zhengruifeng Jul 23, 2019
92c2c86
init
zhengruifeng Jul 23, 2019
c0ba410
add testsuites
zhengruifeng Jul 25, 2019
8b297ff
update transform
zhengruifeng Aug 8, 2019
13027cb
update transform II
zhengruifeng Aug 8, 2019
e46d3a1
nit
zhengruifeng Aug 8, 2019
8709ed1
move predictLeaf in to superclass
zhengruifeng Aug 10, 2019
eda4192
add some comments
zhengruifeng Aug 10, 2019
a4e60e3
add trait TreeEnsembleClassifierParams & TreeEnsembleRegressorParams
zhengruifeng Aug 15, 2019
dd68ac6
rename trait of impurity
zhengruifeng Aug 15, 2019
1d59303
nit
zhengruifeng Aug 15, 2019
3904602
revert trait renaming
zhengruifeng Aug 16, 2019
ae1cb9d
make test suites more common
zhengruifeng Aug 19, 2019
b21f8e8
nit
zhengruifeng Aug 19, 2019
d9d7368
update suites
zhengruifeng Aug 19, 2019
5c5a76e
mv structure comment into the block
zhengruifeng Aug 19, 2019
8247395
update
zhengruifeng Aug 21, 2019
4eb97be
revert setVarianceCol to avoid mima
zhengruifeng Aug 21, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 24 additions & 4 deletions mllib/src/main/scala/org/apache/spark/ml/classification/DecisionTreeClassifier.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,8 +34,8 @@ import org.apache.spark.ml.util.Instrumentation.instrumented
import org.apache.spark.mllib.tree.configuration.{Algo => OldAlgo, Strategy => OldStrategy}
import org.apache.spark.mllib.tree.model.{DecisionTreeModel => OldDecisionTreeModel}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{Dataset, Row}
import org.apache.spark.sql.functions.{col, lit}
import org.apache.spark.sql.{DataFrame, Dataset, Row}
import org.apache.spark.sql.functions.{col, lit, udf}
import org.apache.spark.sql.types.DoubleType

/**
Expand All @@ -55,6 +55,10 @@ class DecisionTreeClassifier @Since("1.4.0") (

// Override parameter setters from parent trait for Java API compatibility.

/** @group setParam */
@Since("3.0.0")
def setLeafCol(value: String): this.type = set(leafCol, value)

@mgaido91 mgaido91 Aug 10, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we can do some refactoring here, in order to dedup this. Can we add it to a trait?

@zhengruifeng zhengruifeng Aug 12, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This way seems like mllib's convention that not add setter into the xxxParam-like trait, like setVarianceCol in DecisionTreeRegressionModel

@srowen srowen Aug 14, 2019

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, agree, it shouldn't be in the model classes. So DecisionTreeClassifierParams doesn't help. Hm... per the discussion below, I agree it's extra weight to refactor the common elements into a superclass of two decision tree classifiers, but it might well be worth it. It looks like it would save a few hundred lines of duplicated code? that would mitigate the concern about the large change here. I'm lightly in favor of going that way. I wouldn't do it for 10 lines of code or something.


/** @group setParam */
@Since("1.4.0")
def setMaxDepth(value: Int): this.type = set(maxDepth, value)
Expand Down Expand Up @@ -136,8 +140,8 @@ class DecisionTreeClassifier @Since("1.4.0") (
val strategy = getOldStrategy(categoricalFeatures, numClasses)
instr.logNumClasses(numClasses)
instr.logParams(this, labelCol, featuresCol, predictionCol, rawPredictionCol,
probabilityCol, maxDepth, maxBins, minInstancesPerNode, minInfoGain, maxMemoryInMB,
cacheNodeIds, checkpointInterval, impurity, seed)
probabilityCol, leafCol, maxDepth, maxBins, minInstancesPerNode, minInfoGain,
maxMemoryInMB, cacheNodeIds, checkpointInterval, impurity, seed)

val trees = RandomForest.run(instances, strategy, numTrees = 1, featureSubsetStrategy = "all",
seed = $(seed), instr = Some(instr), parentUID = Some(uid))
Expand Down Expand Up @@ -210,6 +214,22 @@ class DecisionTreeClassificationModel private[ml] (
rootNode.predictImpl(features).prediction
}

/** @group setParam */
@Since("3.0.0")
def setLeafCol(value: String): this.type = set(leafCol, value)

override def transform(dataset: Dataset[_]): DataFrame = {
transformSchema(dataset.schema, logging = true)

if ($(leafCol).nonEmpty) {
val leafUDF = udf { features: Vector => predictLeaf(features) }
Comment thread
srowen marked this conversation as resolved.
super.transform(dataset)
.withColumn($(leafCol), leafUDF(col($(featuresCol))))
} else {
super.transform(dataset)

@srowen srowen Aug 20, 2019

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's trivial but I guess you could avoid calling this in two places ... call it once and either return it, or the result of it with a new column.

}
}

override protected def predictRaw(features: Vector): Vector = {
Vectors.dense(rootNode.predictImpl(features).impurityStats.stats.clone())
}
Expand Down
26 changes: 23 additions & 3 deletions mllib/src/main/scala/org/apache/spark/ml/classification/GBTClassifier.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,7 @@ import org.apache.spark.ml.util.DefaultParamsReader.Metadata
import org.apache.spark.ml.util.Instrumentation.instrumented
import org.apache.spark.mllib.tree.configuration.{Algo => OldAlgo}
import org.apache.spark.mllib.tree.model.{GradientBoostedTreesModel => OldGBTModel}
import org.apache.spark.sql.{Dataset, Row}
import org.apache.spark.sql.{DataFrame, Dataset, Row}
import org.apache.spark.sql.functions._

/**
Expand Down Expand Up @@ -67,6 +67,10 @@ class GBTClassifier @Since("1.4.0") (

// Parameters from TreeClassifierParams:

/** @group setParam */
@Since("3.0.0")
def setLeafCol(value: String): this.type = set(leafCol, value)

/** @group setParam */
@Since("1.4.0")
def setMaxDepth(value: Int): this.type = set(maxDepth, value)
Expand Down Expand Up @@ -191,8 +195,8 @@ class GBTClassifier @Since("1.4.0") (

instr.logPipelineStage(this)
instr.logDataset(dataset)
instr.logParams(this, labelCol, featuresCol, predictionCol, impurity, lossType,
maxDepth, maxBins, maxIter, maxMemoryInMB, minInfoGain, minInstancesPerNode,
instr.logParams(this, labelCol, featuresCol, predictionCol, leafCol, impurity,
lossType, maxDepth, maxBins, maxIter, maxMemoryInMB, minInfoGain, minInstancesPerNode,
seed, stepSize, subsamplingRate, cacheNodeIds, checkpointInterval, featureSubsetStrategy,
validationIndicatorCol, validationTol)
instr.logNumClasses(numClasses)
Expand Down Expand Up @@ -286,6 +290,22 @@ class GBTClassificationModel private[ml](
@Since("1.4.0")
override def treeWeights: Array[Double] = _treeWeights

/** @group setParam */
@Since("3.0.0")
def setLeafCol(value: String): this.type = set(leafCol, value)

override def transform(dataset: Dataset[_]): DataFrame = {
transformSchema(dataset.schema, logging = true)

if ($(leafCol).nonEmpty) {
val leafUDF = udf { features: Vector => predictLeaf(features) }
super.transform(dataset)
.withColumn($(leafCol), leafUDF(col($(featuresCol))))
} else {
super.transform(dataset)
}
}

override def predict(features: Vector): Double = {
// If thresholds defined, use predictRaw to get probabilities, otherwise use optimization
if (isDefined(thresholds)) {
Expand Down
2 changes: 1 addition & 1 deletion mllib/src/main/scala/org/apache/spark/ml/classification/ProbabilisticClassifier.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@ import org.apache.spark.sql.types.{DataType, StructType}
/**
* (private[classification]) Params for probabilistic classification.
*/
private[classification] trait ProbabilisticClassifierParams
private[ml] trait ProbabilisticClassifierParams
extends ClassifierParams with HasProbabilityCol with HasThresholds {
override protected def validateAndTransformSchema(
schema: StructType,
Expand Down
28 changes: 25 additions & 3 deletions mllib/src/main/scala/org/apache/spark/ml/classification/RandomForestClassifier.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,8 @@ import org.apache.spark.ml.util.Instrumentation.instrumented
import org.apache.spark.mllib.tree.configuration.{Algo => OldAlgo}
import org.apache.spark.mllib.tree.model.{RandomForestModel => OldRandomForestModel}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.Dataset
import org.apache.spark.sql.{DataFrame, Dataset}
import org.apache.spark.sql.functions.{col, udf}

/**
* <a href="http://en.wikipedia.org/wiki/Random_forest">Random Forest</a> learning algorithm for
Expand All @@ -55,6 +56,10 @@ class RandomForestClassifier @Since("1.4.0") (

// Parameters from TreeClassifierParams:

/** @group setParam */
@Since("3.0.0")
def setLeafCol(value: String): this.type = set(leafCol, value)

/** @group setParam */
@Since("1.4.0")
def setMaxDepth(value: Int): this.type = set(maxDepth, value)
Expand Down Expand Up @@ -135,8 +140,9 @@ class RandomForestClassifier @Since("1.4.0") (
super.getOldStrategy(categoricalFeatures, numClasses, OldAlgo.Classification, getOldImpurity)

instr.logParams(this, labelCol, featuresCol, predictionCol, probabilityCol, rawPredictionCol,
impurity, numTrees, featureSubsetStrategy, maxDepth, maxBins, maxMemoryInMB, minInfoGain,
minInstancesPerNode, seed, subsamplingRate, thresholds, cacheNodeIds, checkpointInterval)
leafCol, impurity, numTrees, featureSubsetStrategy, maxDepth, maxBins, maxMemoryInMB,
minInfoGain, minInstancesPerNode, seed, subsamplingRate, thresholds, cacheNodeIds,
checkpointInterval)

val trees = RandomForest
.run(instances, strategy, getNumTrees, getFeatureSubsetStrategy, getSeed, Some(instr))
Expand Down Expand Up @@ -207,6 +213,22 @@ class RandomForestClassificationModel private[ml] (
@Since("1.4.0")
override def treeWeights: Array[Double] = _treeWeights

/** @group setParam */
@Since("3.0.0")
def setLeafCol(value: String): this.type = set(leafCol, value)

override def transform(dataset: Dataset[_]): DataFrame = {
transformSchema(dataset.schema, logging = true)

if ($(leafCol).nonEmpty) {
val leafUDF = udf { features: Vector => predictLeaf(features) }
super.transform(dataset)
.withColumn($(leafCol), leafUDF(col($(featuresCol))))
} else {
super.transform(dataset)
}
}

override protected def predictRaw(features: Vector): Vector = {
// TODO: When we add a generic Bagging class, handle transform there: SPARK-7128
// Classifies using majority votes.
Expand Down
17 changes: 12 additions & 5 deletions mllib/src/main/scala/org/apache/spark/ml/regression/DecisionTreeRegressor.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -209,28 +209,35 @@ class DecisionTreeRegressionModel private[ml] (
rootNode.predictImpl(features).impurityStats.calculate()
}

/** @group setParam */
@Since("3.0.0")
def setLeafCol(value: String): this.type = set(leafCol, value)

@Since("2.0.0")
override def transform(dataset: Dataset[_]): DataFrame = {
transformSchema(dataset.schema, logging = true)
transformImpl(dataset)
}

override protected def transformImpl(dataset: Dataset[_]): DataFrame = {
var predictionColNames = Seq.empty[String]
var predictionColumns = Seq.empty[Column]

if ($(predictionCol).nonEmpty) {
val predictUDF = udf { (features: Vector) => predict(features) }
val predictUDF = udf { features: Vector => predict(features) }
predictionColNames :+= $(predictionCol)
predictionColumns :+= predictUDF(col($(featuresCol)))
}

if (isDefined(varianceCol) && $(varianceCol).nonEmpty) {
val predictVarianceUDF = udf { (features: Vector) => predictVariance(features) }
val predictVarianceUDF = udf { features: Vector => predictVariance(features) }
predictionColNames :+= $(varianceCol)
predictionColumns :+= predictVarianceUDF(col($(featuresCol)))
}

if ($(leafCol).nonEmpty) {
val leafUDF = udf { features: Vector => predictLeaf(features) }
predictionColNames :+= $(leafCol)
predictionColumns :+= leafUDF(col($(featuresCol)))
}

if (predictionColNames.nonEmpty) {
dataset.withColumns(predictionColNames, predictionColumns)
} else {
Expand Down
41 changes: 35 additions & 6 deletions mllib/src/main/scala/org/apache/spark/ml/regression/GBTRegressor.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,7 @@ import org.apache.spark.ml.util.DefaultParamsReader.Metadata
import org.apache.spark.ml.util.Instrumentation.instrumented
import org.apache.spark.mllib.tree.configuration.{Algo => OldAlgo}
import org.apache.spark.mllib.tree.model.{GradientBoostedTreesModel => OldGBTModel}
import org.apache.spark.sql.{DataFrame, Dataset, Row}
import org.apache.spark.sql.{Column, DataFrame, Dataset, Row}
import org.apache.spark.sql.functions._

/**
Expand Down Expand Up @@ -66,6 +66,10 @@ class GBTRegressor @Since("1.4.0") (@Since("1.4.0") override val uid: String)

// Parameters from TreeRegressorParams:

/** @group setParam */
@Since("3.0.0")
def setLeafCol(value: String): this.type = set(leafCol, value)

/** @group setParam */
@Since("1.4.0")
def setMaxDepth(value: Int): this.type = set(maxDepth, value)
Expand Down Expand Up @@ -169,7 +173,7 @@ class GBTRegressor @Since("1.4.0") (@Since("1.4.0") override val uid: String)

instr.logPipelineStage(this)
instr.logDataset(dataset)
instr.logParams(this, labelCol, featuresCol, predictionCol, impurity, lossType,
instr.logParams(this, labelCol, featuresCol, predictionCol, leafCol, impurity, lossType,
maxDepth, maxBins, maxIter, maxMemoryInMB, minInfoGain, minInstancesPerNode,
seed, stepSize, subsamplingRate, cacheNodeIds, checkpointInterval, featureSubsetStrategy,
validationIndicatorCol, validationTol)
Expand Down Expand Up @@ -245,12 +249,37 @@ class GBTRegressionModel private[ml](
@Since("1.4.0")
override def treeWeights: Array[Double] = _treeWeights

override protected def transformImpl(dataset: Dataset[_]): DataFrame = {
/** @group setParam */
@Since("3.0.0")
def setLeafCol(value: String): this.type = set(leafCol, value)

override def transform(dataset: Dataset[_]): DataFrame = {
transformSchema(dataset.schema, logging = true)

var predictionColNames = Seq.empty[String]
var predictionColumns = Seq.empty[Column]

val bcastModel = dataset.sparkSession.sparkContext.broadcast(this)
val predictUDF = udf { (features: Any) =>
bcastModel.value.predict(features.asInstanceOf[Vector])

if ($(predictionCol).nonEmpty) {
val predictUDF = udf { features: Vector => bcastModel.value.predict(features) }
predictionColNames :+= $(predictionCol)
predictionColumns :+= predictUDF(col($(featuresCol)))
}

if ($(leafCol).nonEmpty) {
val leafUDF = udf { features: Vector => bcastModel.value.predictLeaf(features) }
predictionColNames :+= $(leafCol)
predictionColumns :+= leafUDF(col($(featuresCol)))
}

if (predictionColNames.nonEmpty) {
dataset.withColumns(predictionColNames, predictionColumns)
} else {
this.logWarning(s"$uid: GBTRegressionModel.transform() does nothing" +
" because no output columns were set.")
dataset.toDF()
}
dataset.withColumn($(predictionCol), predictUDF(col($(featuresCol))))
}

override def predict(features: Vector): Double = {
Expand Down
41 changes: 35 additions & 6 deletions mllib/src/main/scala/org/apache/spark/ml/regression/RandomForestRegressor.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,7 @@ import org.apache.spark.ml.util.DefaultParamsReader.Metadata
import org.apache.spark.ml.util.Instrumentation.instrumented
import org.apache.spark.mllib.tree.configuration.{Algo => OldAlgo}
import org.apache.spark.mllib.tree.model.{RandomForestModel => OldRandomForestModel}
import org.apache.spark.sql.{DataFrame, Dataset}
import org.apache.spark.sql.{Column, DataFrame, Dataset}
import org.apache.spark.sql.functions.{col, udf}

/**
Expand All @@ -51,6 +51,10 @@ class RandomForestRegressor @Since("1.4.0") (@Since("1.4.0") override val uid: S

// Parameters from TreeRegressorParams:

/** @group setParam */
@Since("3.0.0")
def setLeafCol(value: String): this.type = set(leafCol, value)

/** @group setParam */
@Since("1.4.0")
def setMaxDepth(value: Int): this.type = set(maxDepth, value)
Expand Down Expand Up @@ -123,7 +127,7 @@ class RandomForestRegressor @Since("1.4.0") (@Since("1.4.0") override val uid: S

instr.logPipelineStage(this)
instr.logDataset(instances)
instr.logParams(this, labelCol, featuresCol, predictionCol, impurity, numTrees,
instr.logParams(this, labelCol, featuresCol, predictionCol, leafCol, impurity, numTrees,
featureSubsetStrategy, maxDepth, maxBins, maxMemoryInMB, minInfoGain,
minInstancesPerNode, seed, subsamplingRate, cacheNodeIds, checkpointInterval)

Expand Down Expand Up @@ -191,12 +195,37 @@ class RandomForestRegressionModel private[ml] (
@Since("1.4.0")
override def treeWeights: Array[Double] = _treeWeights

override protected def transformImpl(dataset: Dataset[_]): DataFrame = {
/** @group setParam */
@Since("3.0.0")
def setLeafCol(value: String): this.type = set(leafCol, value)

override def transform(dataset: Dataset[_]): DataFrame = {
transformSchema(dataset.schema, logging = true)

var predictionColNames = Seq.empty[String]
var predictionColumns = Seq.empty[Column]

val bcastModel = dataset.sparkSession.sparkContext.broadcast(this)
val predictUDF = udf { (features: Any) =>
bcastModel.value.predict(features.asInstanceOf[Vector])

if ($(predictionCol).nonEmpty) {
val predictUDF = udf { features: Vector => bcastModel.value.predict(features) }
predictionColNames :+= $(predictionCol)
predictionColumns :+= predictUDF(col($(featuresCol)))
}

if ($(leafCol).nonEmpty) {
val leafUDF = udf { features: Vector => bcastModel.value.predictLeaf(features) }
predictionColNames :+= $(leafCol)
predictionColumns :+= leafUDF(col($(featuresCol)))
}

if (predictionColNames.nonEmpty) {
dataset.withColumns(predictionColNames, predictionColumns)
} else {
this.logWarning(s"$uid: RandomForestRegressionModel.transform() does nothing" +
" because no output columns were set.")
dataset.toDF()
}
dataset.withColumn($(predictionCol), predictUDF(col($(featuresCol))))
}

override def predict(features: Vector): Double = {
Expand Down
Loading