[SPARK-7127] Replaced deprecated callUDF with udf

Author: BryanCutler

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

@@ -181,8 +181,10 @@ final class GBTClassificationModel(
 
   override protected def transformImpl(dataset: DataFrame): DataFrame = {
     val bcastModel = dataset.sqlContext.sparkContext.broadcast(this)
-    dataset.withColumn($(predictionCol), callUDF(bcastModel.value.predict _, DoubleType,
-      col($(featuresCol))))
+    val predictUDF = udf { (features: Any) =>
+      bcastModel.value.predict(features.asInstanceOf[Vector])
+    }
+    dataset.withColumn($(predictionCol), predictUDF(col($(featuresCol))))
   }
 
   override protected def predict(features: Vector): Double = {

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

@@ -139,8 +139,10 @@ final class RandomForestClassificationModel private[ml] (
 
   override protected def transformImpl(dataset: DataFrame): DataFrame = {
     val bcastModel = dataset.sqlContext.sparkContext.broadcast(this)
-    dataset.withColumn($(predictionCol), callUDF(bcastModel.value.predict _, DoubleType,
-      col($(featuresCol))))
+    val predictUDF = udf { (features: Any) =>
+      bcastModel.value.predict(features.asInstanceOf[Vector])
+    }
+    dataset.withColumn($(predictionCol), predictUDF(col($(featuresCol))))
   }
 
   override protected def predict(features: Vector): Double = {

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

@@ -171,8 +171,10 @@ final class GBTRegressionModel(
 
   override protected def transformImpl(dataset: DataFrame): DataFrame = {
     val bcastModel = dataset.sqlContext.sparkContext.broadcast(this)
-    dataset.withColumn($(predictionCol), callUDF(bcastModel.value.predict _, DoubleType,
-      col($(featuresCol))))
+    val predictUDF = udf { (features: Any) =>
+      bcastModel.value.predict(features.asInstanceOf[Vector])
+    }
+    dataset.withColumn($(predictionCol), predictUDF(col($(featuresCol))))
   }
 
   override protected def predict(features: Vector): Double = {

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

@@ -126,8 +126,10 @@ final class RandomForestRegressionModel private[ml] (
 
   override protected def transformImpl(dataset: DataFrame): DataFrame = {
     val bcastModel = dataset.sqlContext.sparkContext.broadcast(this)
-    dataset.withColumn($(predictionCol), callUDF(bcastModel.value.predict _, DoubleType,
-      col($(featuresCol))))
+    val predictUDF = udf { (features: Any) =>
+      bcastModel.value.predict(features.asInstanceOf[Vector])
+    }
+    dataset.withColumn($(predictionCol), predictUDF(col($(featuresCol))))
   }
 
   override protected def predict(features: Vector): Double = {