Skip to content

[SPARK-7318][Streaming] DStream cleans objects that are not closures #5860

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
andrewor14 wants to merge 5 commits into from
Closed

[SPARK-7318][Streaming] DStream cleans objects that are not closures #5860

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
a4fa768
Clean the closure, not the RDD
May 2, 2015
67eeff4
Add tests
May 2, 2015
eed3390
Merge branch 'master' of github.com:apache/spark into streaming-closu…
May 3, 2015
5ee4e25
Fix tests
May 3, 2015
8e971d7
Do not throw exception if object to clean is not closure
May 3, 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
5 changes: 5 additions & 0 deletions core/src/main/scala/org/apache/spark/util/ClosureCleaner.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -179,6 +179,11 @@ private[spark] object ClosureCleaner extends Logging {
cleanTransitively: Boolean,
accessedFields: Map[Class[_], Set[String]]): Unit = {

if (!isClosure(func.getClass)) {
logWarning("Expected a closure; got " + func.getClass.getName)
Copy link
Contributor

@tdas tdas May 5, 2015

Choose a reason for hiding this comment

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

@pwendell Is this okay to log a warning?

Copy link
Contributor

@pwendell pwendell May 5, 2015

Choose a reason for hiding this comment

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

Any reason not to make this an assertion? Isn't it simply invalid if we call this on something that is not a closure?

return
}

// TODO: clean all inner closures first. This requires us to find the inner objects.
// TODO: cache outerClasses / innerClasses / accessedFields

Expand Down
3 changes: 2 additions & 1 deletion streaming/src/main/scala/org/apache/spark/streaming/dstream/DStream.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -553,7 +553,8 @@ abstract class DStream[T: ClassTag] (
// because the DStream is reachable from the outer object here, and because
// DStreams can't be serialized with closures, we can't proactively check
// it for serializability and so we pass the optional false to SparkContext.clean
transform((r: RDD[T], t: Time) => context.sparkContext.clean(transformFunc(r), false))
val cleanedF = context.sparkContext.clean(transformFunc, false)
transform((r: RDD[T], t: Time) => cleanedF(r))
}

/**
Expand Down
4 changes: 2 additions & 2 deletions streaming/src/test/scala/org/apache/spark/streaming/ReceiverSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -256,8 +256,8 @@ class ReceiverSuite extends TestSuiteBase with Timeouts with Serializable {
}

withStreamingContext(new StreamingContext(sparkConf, batchDuration)) { ssc =>
val receiver1 = ssc.sparkContext.clean(new FakeReceiver(sendData = true))
val receiver2 = ssc.sparkContext.clean(new FakeReceiver(sendData = true))
val receiver1 = new FakeReceiver(sendData = true)
val receiver2 = new FakeReceiver(sendData = true)
val receiverStream1 = ssc.receiverStream(receiver1)
val receiverStream2 = ssc.receiverStream(receiver2)
receiverStream1.register()
Expand Down