Skip to content

[SPARK-7127] [MLLIB] Adding broadcast of model before prediction for ensembles #6300

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
BryanCutler wants to merge 8 commits into from
Closed

[SPARK-7127] [MLLIB] Adding broadcast of model before prediction for ensembles #6300

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
83904bb
[SPARK-7127] Adding broadcast of model before prediction in RandomFor…
BryanCutler May 20, 2015
aaad77b
[SPARK-7127] Removed abstract class for broadcasting model, instead p…
BryanCutler May 25, 2015
6fd153c
[SPARK-7127] Applied broadcasting to remaining ensemble models
BryanCutler Jun 15, 2015
171a6ce
[SPARK-7127] Used modelAccessor parameter in predictImpl to access br…
BryanCutler Jun 15, 2015
1f34be4
[SPARK-7127] Removed accidental newline
BryanCutler Jun 15, 2015
9afad56
[SPARK-7127] Simplified calls by overriding transformImpl and using b…
BryanCutler Jul 14, 2015
40a139d
Merge branch 'master' into bcast-ensemble-models-7127
BryanCutler Jul 15, 2015
86e73de
[SPARK-7127] Replaced deprecated callUDF with udf
BryanCutler Jul 15, 2015
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
8 changes: 7 additions & 1 deletion mllib/src/main/scala/org/apache/spark/ml/Predictor.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -174,9 +174,15 @@ abstract class PredictionModel[FeaturesType, M <: PredictionModel[FeaturesType,
* @return transformed dataset with [[predictionCol]] of type [[Double]]
*/
override def transform(dataset: DataFrame): DataFrame = {
transformImpl(dataset, predict)
}

@jkbradley jkbradley Jul 9, 2015

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.

This could call:

transformImpl(dataset)

protected def transformImpl(
dataset: DataFrame,
predictFunc: (FeaturesType) => Double): DataFrame = {
transformSchema(dataset.schema, logging = true)
if ($(predictionCol).nonEmpty) {
dataset.withColumn($(predictionCol), callUDF(predict _, DoubleType, col($(featuresCol))))
dataset.withColumn($(predictionCol), callUDF(predictFunc, DoubleType, col($(featuresCol))))
} else {
this.logWarning(s"$uid: Predictor.transform() was called as NOOP" +
" since no output columns were set.")
Expand Down
15 changes: 13 additions & 2 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 @@ -109,6 +109,7 @@ object RandomForestClassifier {
RandomForestParams.supportedFeatureSubsetStrategies
}


/**
* :: AlphaComponent ::
*
Expand All @@ -134,13 +135,23 @@ final class RandomForestClassificationModel private[ml] (

override def treeWeights: Array[Double] = _treeWeights

override def transform(dataset: DataFrame): DataFrame = {
val bcastModel = dataset.sqlContext.sparkContext.broadcast(this)

@BryanCutler BryanCutler May 25, 2015

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

You mentioned that we might want to selectively broadcast the model, only if it's large enough. Do you think that is something we can do here automatically, or would it need to be a configuration setting?

val predictFunc = (features: Vector) => predictImpl(features, () => bcastModel.value)
transformImpl(dataset, predictFunc)
}

override protected def predict(features: Vector): Double = {
// TODO: Override transform() to broadcast model. SPARK-7127
// TODO: When we add a generic Bagging class, handle transform there: SPARK-7128
// Predict without using a broadcasted model
predictImpl(features, () => this)

@BryanCutler BryanCutler May 20, 2015

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Do we still want to support prediction without broadcasting?

@jkbradley jkbradley May 21, 2015

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, we will want to. If it's a small model, then it can be faster to not broadcast it.

}

protected def predictImpl(features: Vector, modelAccesor: () => TreeEnsembleModel): Double = {
// Classifies using majority votes.
// Ignore the weights since all are 1.0 for now.
val votes = mutable.Map.empty[Int, Double]
_trees.view.foreach { tree =>
modelAccesor().trees.view.foreach { tree =>
val prediction = tree.rootNode.predict(features).toInt
votes(prediction) = votes.getOrElse(prediction, 0.0) + 1.0 // 1.0 = weight
}
Expand Down