[SPARK-7741] [STREAMING] Added sparkContext.clean at missing places

[tdas]

These missing cleaners lead to capturing unnecessary objects in closures, especially when using a spark streaming in spark-shell.

[tdas]

@andrewor14 Could you please take a look?

[AmplabJenkins]

Merged build triggered.

[AmplabJenkins]

Merged build started.

[SparkQA]

Test build #33096 has started for PR 6268 at commit c1dc4cf.

[andrewor14]

This one's actually not necessary because it will be cleaned by reduceByKey later

[andrewor14]

@tdas Most of the changes are actually not required, because they eventually get cleaned later through other methods. We established this rule in #5787.

So the only places that are actually required are: filter, combineByKey, mapValues, flatMapValues, and foreachRDD.

Additionally there are a few calls in updateStateByKey that need to be cleaned. In general, the following pattern needs cleaning:

def somePublicMethod(func: () => Int): Unit = {
  val cleanedF = sc.clean(func)
  new SomeRDD(this, (unused: Int) => cleanedF())
}

I've opened #6269 to describe what I mean.

[SparkQA]

Test build #33096 has finished for PR 6268 at commit c1dc4cf.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[AmplabJenkins]

Merged build finished. Test PASSed.

[AmplabJenkins]

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/33096/
Test PASSed.

[tdas]

@andrewor14 I am closing this PR, yours is far better :)