[SPARK-2373]RDD add span function (split an RDD to two RDD based on user's function)]

[YanjieGao]

Hi all,
This function is a basic function in Scala.
def span(p: T => Boolean): (RDD[T], RDD[T])
Splits this RDD into a prefix/suffix pair according to a predicate .
returns
a pair consisting of the longest prefix of this RDD whose elements all satisfy p, and the rest of this list.

JIRA:https://issues.apache.org/jira/browse/SPARK-2373

[YanjieGao]

Thanks ,I optimize the code so it only evaluates the function once .Other comments are on JIRA

[YanjieGao]

This function is useful in some cases ,Such as when i do Skew Join in another PR,I need to split an RDD to two RDD,One has skew keys ,and the other is not .
val (maxKeySkewedTable, mainSkewedTable) = skewedTable.span(row => {
skewSideKeyGenerator(row).toString().equals(maxrowKey.toString())
})

[andrewor14]

test this please

[SparkQA]

QA tests have started for PR 1306 at commit fab7ed9.

• This patch merges cleanly.

[SparkQA]

QA tests have finished for PR 1306 at commit fab7ed9.

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

[JoshRosen]

This needs scaladoc, since this method's meaning won't be clear to users unless they're familiar with the Scala span function.

[JoshRosen]

@mateiz Thoughts on this? Do you think this API will be broadly useful enough to include?

[mateiz]

IMO this is too specialized to include. It's small enough that applications can do it themselves, but also fairly confusing unless your RDD is already sorted in some way. I think we should just leave it for applications to do it. If you are doing a skewed join operator for example, you can do it within the implementation of that but not show it to the user.

[YanjieGao]

Ok ,Got it, I will close this PR ;

[crakjie]

It's a nice start to code the scala def partition(p: (A) ⇒ Boolean): (List[A], List[A]) thank.
But it's better to name it fork to avoid confusion,

[k0ala]

FWIW, I frequently have this use case and would love a span on RDD.