Do not forget foreachRDD

Andrew Or · Andrew Or · commit 328139b1b223 · 2015-05-19T16:58:00.000-07:00
diff --git a/streaming/src/main/scala/org/apache/spark/streaming/dstream/DStream.scala b/streaming/src/main/scala/org/apache/spark/streaming/dstream/DStream.scala
@@ -624,7 +624,8 @@ abstract class DStream[T: ClassTag] (
    * 'this' DStream will be registered as an output stream and therefore materialized.
    */
   def foreachRDD(foreachFunc: RDD[T] => Unit): Unit = ssc.withScope {
-    this.foreachRDD((r: RDD[T], t: Time) => foreachFunc(r))
+    val cleanedF = context.sparkContext.clean(foreachFunc)
+    this.foreachRDD((r: RDD[T], t: Time) => cleanedF(r))
   }
 
   /**

Original file line number	Diff line number	Diff line change
`@@ -624,7 +624,8 @@ abstract class DStream[T: ClassTag] (`
`624`	`624`	`* 'this' DStream will be registered as an output stream and therefore materialized.`
`625`	`625`	`*/`
`626`	`626`	`def foreachRDD(foreachFunc: RDD[T] => Unit): Unit = ssc.withScope {`
`627`		`- this.foreachRDD((r: RDD[T], t: Time) => foreachFunc(r))`
	`627`	`+ val cleanedF = context.sparkContext.clean(foreachFunc)`
	`628`	`+ this.foreachRDD((r: RDD[T], t: Time) => cleanedF(r))`
`628`	`629`	`}`
`629`	`630`
`630`	`631`	`/**`