[SPARK-4397][Core] Reorganize 'implicit's to improve the API convenience

[zsxwing]

This PR moved implicits to package object and companion object to enable the Scala compiler search them automatically without explicit importing.

It should not break any API. A test project for backforward compatibility is here. It proves the codes compiled with Spark 1.1.0 can run with this PR.

To summarize, the changes are:

• Deprecated the old implicit conversion functions: this preserves binary compatibility for code compiled against earlier versions of Spark.
• Removed "implicit" from them so they are just normal functions: this made sure the compiler doesn't get confused and warn about multiple implicits in scope.
• Created new implicit functions in package rdd object, which is part of the scope that scalac will search when looking for implicit conversions on various RDD objects.

The disadvantage is there are duplicated codes in SparkContext for backforward compatibility.

[zsxwing]

/cc @rxin

[SparkQA]

Test build #23354 has started for PR 3262 at commit 1eda9e4.

• This patch merges cleanly.

[SparkQA]

Test build #23354 has finished for PR 3262 at commit 1eda9e4.

• This patch fails RAT tests.
• This patch merges cleanly.
• This patch adds no public classes.

[AmplabJenkins]

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

[SparkQA]

Test build #23356 has started for PR 3262 at commit 3ac4f07.

• This patch merges cleanly.

[rxin]

can u add parentheses to groupByKey?

[SparkQA]

Test build #23358 has started for PR 3262 at commit 9b73188.

• This patch merges cleanly.

[aarondav]

What's the distinction for intToIntWritable/writableConverters?

[SparkQA]

Test build #23356 has finished for PR 3262 at commit 3ac4f07.

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

[AmplabJenkins]

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

[SparkQA]

Test build #23358 has finished for PR 3262 at commit 9b73188.

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

[AmplabJenkins]

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

[SparkQA]

Test build #23367 has started for PR 3262 at commit 3bdcae2.

• This patch merges cleanly.

[zsxwing]

What's the distinction for intToIntWritable/writableConverters?

writableConverters can work. Already done. Here is the code to test binary compatibility.

import org.apache.spark.{SparkContext, SparkConf}
import org.apache.spark.SparkContext._

object ImplicitBackforwardCompatibilityApp {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("ImplicitBackforwardCompatibilityApp")
    val sc = new SparkContext(conf)

    val rdd = sc.parallelize(1 to 100).map(i => (i, i))
    val rdd2 = rdd.groupByKey() // rddToPairRDDFunctions
    val rdd3 = rdd2.sortByKey() // rddToOrderedRDDFunctions
    val s1 = rdd3.map(_._1).stats() // numericRDDToDoubleRDDFunctions
    println(s1)
    val s2 = rdd3.map(_._1.toDouble).stats() // doubleRDDToDoubleRDDFunctions
    println(s2)
    val f = rdd2.countAsync() // rddToAsyncRDDActions
    println(f.get())
    rdd2.map { case (k, v) => (k, v.size)} saveAsSequenceFile ("/tmp/implicit_test_path") // rddToSequenceFileRDDFunctions

    val a1 = sc.accumulator(123.4) // DoubleAccumulatorParam
    a1.add(1.0)
    println(a1.value)
    val a2 = sc.accumulator(123) // IntAccumulatorParam
    a2.add(3)
    println(a2.value)
    val a3 = sc.accumulator(123L) // LongAccumulatorParam
    a3.add(11L)
    println(a3.value)
    val a4 = sc.accumulator(123F) // FloatAccumulatorParam
    a4.add(1.1F)
    println(a4.value)

    {
      sc.parallelize(1 to 10).map(i => (i, i)).saveAsSequenceFile("/tmp/implicit_test_int")
      val r = sc.sequenceFile[Int, Int]("/tmp/implicit_test_int")
      r.map { case (k, v) => (k.toString, v.toString)} foreach (println)
    }

    {
      sc.parallelize(1 to 10).map(i => (i.toLong, i.toLong)).saveAsSequenceFile("/tmp/implicit_test_long")
      val r = sc.sequenceFile[Long, Long]("/tmp/implicit_test_long")
      r.map { case (k, v) => (k.toString, v.toString)} foreach (println)
    }

    {
      sc.parallelize(1 to 10).map(i => (i.toDouble, i.toDouble)).saveAsSequenceFile("/tmp/implicit_test_double")
      val r = sc.sequenceFile[Double, Double]("/tmp/implicit_test_double")
      r.map { case (k, v) => (k.toString, v.toString)} foreach (println)
    }

    {
      sc.parallelize(1 to 10).map(i => (i.toFloat, i.toFloat)).saveAsSequenceFile("/tmp/implicit_test_float")
      val r = sc.sequenceFile[Float, Float]("/tmp/implicit_test_float")
      r.map { case (k, v) => (k.toString, v.toString)} foreach (println)
    }

    {
      sc.parallelize(1 to 10).map(i => (i.toString, i.toString)).saveAsSequenceFile("/tmp/implicit_test_string")
      val r = sc.sequenceFile[String, String]("/tmp/implicit_test_string")
      r.map { case (k, v) => (k.toString, v.toString)} foreach (println)
    }

    {
      sc.parallelize(1 to 10).map(i => (true, false)).saveAsSequenceFile("/tmp/implicit_test_boolean")
      val r = sc.sequenceFile[Boolean, Boolean]("/tmp/implicit_test_boolean")
      r.map { case (k, v) => (k.toString, v.toString)} foreach (println)
    }

    {
      sc.parallelize(1 to 10).map(i => (Array(i.toByte), Array(i.toByte))).saveAsSequenceFile("/tmp/implicit_test_bytes")
      val r = sc.sequenceFile[Array[Byte], Array[Byte]]("/tmp/implicit_test_bytes")
      r.map { case (k, v) => (k.toString, v.toString)} foreach (println)
    }

    {
      sc.parallelize(1 to 10).map(i => (i.toString, i.toString)).saveAsSequenceFile("/tmp/implicit_test_writable")
      val r = sc.sequenceFile[org.apache.hadoop.io.Text, org.apache.hadoop.io.Text]("/tmp/implicit_test_writable")
      r.map { case (k, v) => (k.toString, v.toString)} foreach (println)
    }

    sc.stop()
  }
}

I compiled the above codes with Spark 1.1.0 and ran it with the new Spark compiled from this PR. And it works correctly.

For intToIntWritable, the problem is that the implicit value for SequenceFileRDDFunctions is a function T => Writable[T]. However, we cannot add these xxxToXXXWritable methods to the implicit scope of T => Writable[T] which is out of the Spark's codes. The definition of implicit scope is:

implicit scope, which contains all sort of companion objects and package object that bear some relation to the implicit's type which we search for (i.e. package object of the type, companion object of the type itself, of its type constructor if any, of its parameters if any, and also of its supertype and supertraits).

Ref: http://eed3si9n.com/revisiting-implicits-without-import-tax

A possible solution is creating a new class for T => Writable[T] like WritableConverter, and change the implicit type of SequenceFileRDDFunctions to this class. E.g.

class SequenceFileRDDFunctions[K, V](
    self: RDD[(K, V)])(implicit keyConverter: NewWritableConverter[K], valueConverter: NewWritableConverter[V])

However, since it's a breaking change (of cause, we can also add a new SequenceFileRDDFunctions class to avoid breaking the old codes), I don't think it's worth us to change it.

[SparkQA]

Test build #23367 has finished for PR 3262 at commit 3bdcae2.

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

[AmplabJenkins]

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

[rxin]

update these accordingly too

[mateiz]

@zsxwing just curious, with the old conversions being deprecated, is there any chance they'll create compiler warnings in common uses of the code? In any case this seems pretty cool if it doesn't actually break binary compatibility. I guess one risk is if it adds new implicits that cause something to compile differently, but it seems unlikely from a first glance.

[rxin]

Ok I finally went through the code. I like the change and it is pretty clever. I believe it should preserve both source compatibility and binary compatibility.

To summarize, the changes are:

1. Deprecated the old implicit conversion functions: this preserves binary compatibility for code compiled against earlier versions of Spark.
2. Removed "implicit" from them so they are just normal functions: this made sure the compiler doesn't get confused and warn about multiple implicits in scope.
3. Created new implicit functions in package rdd object, which is part of the scope that scalac will search when looking for implicit conversions on various RDD objects.

It is still a tricky change so it'd be great to get more eyes.

[SparkQA]

Test build #23425 has started for PR 3262 at commit 7266218.

• This patch merges cleanly.

[zsxwing]

@rxin Thank you for the great summary and reviewing. Already updated it accordingly.

[AmplabJenkins]

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

[rxin]

@heathermiller @gzm0 - do you think this pr is good for merge now?

[gzm0]

All these comments are outdated (still refer to package object, but should refer to RDD companion)

[zsxwing]

Thank you. Fixed it.

[gzm0]

Otherwise LGTM

[gzm0]

LGTM

[SparkQA]

Test build #23716 has started for PR 3262 at commit fc30314.

• This patch merges cleanly.

[SparkQA]

Test build #23716 has finished for PR 3262 at commit fc30314.

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

[AmplabJenkins]

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

[rxin]

I'm merging this in master. Thanks for working on this @zsxwing and everybody else for reviewing.

[rxin]

cc @mateiz @pwendell I'm leaving this out of branch-1.2 thinking it is too last minute to merge something like this. Let me know if you want to cherry pick this into branch-1.2.

[mateiz]

Yeah merging to master sounds fine; it's too late to put it in 1.2.

[mateiz]

Thanks for the patch @zsxwing, this is very cool.