use mapValues

mengxr · mengxr · commit 2821b3b7d9ca · 2014-08-18T16:01:23.000-07:00
diff --git a/mllib/src/main/scala/org/apache/spark/mllib/regression/StreamingLinearAlgorithm.scala b/mllib/src/main/scala/org/apache/spark/mllib/regression/StreamingLinearAlgorithm.scala
@@ -20,6 +20,7 @@ package org.apache.spark.mllib.regression
 import org.apache.spark.Logging
 import org.apache.spark.annotation.DeveloperApi
 import org.apache.spark.mllib.linalg.Vector
+import org.apache.spark.streaming.StreamingContext._
 import org.apache.spark.streaming.dstream.DStream
 
 /**
@@ -117,6 +118,6 @@ abstract class StreamingLinearAlgorithm[
       logError(msg)
       throw new IllegalArgumentException(msg)
     }
-    data.mapPartitions(_.map(x => (x._1, model.predict(x._2))), preservePartitioning = true)
+    data.mapValues(model.predict)
   }
 }

Original file line number	Diff line number	Diff line change
`@@ -20,6 +20,7 @@ package org.apache.spark.mllib.regression`
`20`	`20`	`import org.apache.spark.Logging`
`21`	`21`	`import org.apache.spark.annotation.DeveloperApi`
`22`	`22`	`import org.apache.spark.mllib.linalg.Vector`
	`23`	`+import org.apache.spark.streaming.StreamingContext._`
`23`	`24`	`import org.apache.spark.streaming.dstream.DStream`
`24`	`25`
`25`	`26`	`/**`
`@@ -117,6 +118,6 @@ abstract class StreamingLinearAlgorithm[`
`117`	`118`	`logError(msg)`
`118`	`119`	`throw new IllegalArgumentException(msg)`
`119`	`120`	`}`
`120`		`- data.mapPartitions(_.map(x => (x._1, model.predict(x._2))), preservePartitioning = true)`
	`121`	`+ data.mapValues(model.predict)`
`121`	`122`	`}`
`122`	`123`	`}`