diff options
Diffstat (limited to 'repl/src')
-rw-r--r-- | repl/src/main/scala/spark/repl/SparkILoop.scala | 5 | ||||
-rw-r--r-- | repl/src/test/scala/spark/repl/ReplSuiteMixin.scala | 12 | ||||
-rw-r--r-- | repl/src/test/scala/spark/repl/StandaloneClusterReplSuite.scala | 79 |
3 files changed, 84 insertions, 12 deletions
diff --git a/repl/src/main/scala/spark/repl/SparkILoop.scala b/repl/src/main/scala/spark/repl/SparkILoop.scala index 28a7c161f6..5a1e54c929 100644 --- a/repl/src/main/scala/spark/repl/SparkILoop.scala +++ b/repl/src/main/scala/spark/repl/SparkILoop.scala @@ -152,10 +152,7 @@ class SparkILoop(in0: Option[BufferedReader], protected val out: JPrintWriter, finally in = saved } - /*PRASHANT:Detecting if a lazy val has been materialized or not is possible but not worth it - * as in most cases of spark shell usages they will be. Incase they are not user will find - * shutdown slower than the shell start up itself - * */ + def sparkCleanUp(){ echo("Stopping spark context.") intp.beQuietDuring { diff --git a/repl/src/test/scala/spark/repl/ReplSuiteMixin.scala b/repl/src/test/scala/spark/repl/ReplSuiteMixin.scala index 35429bf01f..fd1a1b1e7c 100644 --- a/repl/src/test/scala/spark/repl/ReplSuiteMixin.scala +++ b/repl/src/test/scala/spark/repl/ReplSuiteMixin.scala @@ -14,12 +14,15 @@ import spark.deploy.master.Master import spark.deploy.worker.Worker trait ReplSuiteMixin { + val localIp = "127.0.1.2" + val port = "7089" + val sparkUrl = s"spark://$localIp:$port" def setupStandaloneCluster() { - future { Master.main(Array("-i", "127.0.1.2", "-p", "7089")) } + future { Master.main(Array("-i", localIp, "-p", port, "--webui-port", "0")) } Thread.sleep(2000) - future { Worker.main(Array("spark://127.0.1.2:7089", "--webui-port", "0")) } + future { Worker.main(Array(sparkUrl, "--webui-port", "0")) } } - + def runInterpreter(master: String, input: String): String = { val in = new BufferedReader(new StringReader(input + "\n")) val out = new StringWriter() @@ -33,6 +36,7 @@ trait ReplSuiteMixin { } } } + val interp = new SparkILoop(in, new PrintWriter(out), master) spark.repl.Main.interp = interp val separator = System.getProperty("path.separator") @@ -53,4 +57,4 @@ trait ReplSuiteMixin { assert(!(output contains message), "Interpreter output contained '" + message + "':\n" + output) } -}
\ No newline at end of file +} diff --git a/repl/src/test/scala/spark/repl/StandaloneClusterReplSuite.scala b/repl/src/test/scala/spark/repl/StandaloneClusterReplSuite.scala index a0940e2166..0822770fe2 100644 --- a/repl/src/test/scala/spark/repl/StandaloneClusterReplSuite.scala +++ b/repl/src/test/scala/spark/repl/StandaloneClusterReplSuite.scala @@ -1,12 +1,16 @@ package spark.repl +import java.io.FileWriter + import org.scalatest.FunSuite +import com.google.common.io.Files + class StandaloneClusterReplSuite extends FunSuite with ReplSuiteMixin { setupStandaloneCluster test("simple collect") { - val output = runInterpreter("spark://127.0.1.2:7089", """ + val output = runInterpreter(sparkUrl, """ var x = 123 val data = sc.parallelize(1 to 3).map(_ + x) data.take(3) @@ -17,9 +21,9 @@ class StandaloneClusterReplSuite extends FunSuite with ReplSuiteMixin { assertContains("125", output) assertContains("126", output) } - + test("simple foreach with accumulator") { - val output = runInterpreter("spark://127.0.1.2:7089", """ + val output = runInterpreter(sparkUrl, """ val accum = sc.accumulator(0) sc.parallelize(1 to 10).foreach(x => accum += x) accum.value @@ -29,4 +33,71 @@ class StandaloneClusterReplSuite extends FunSuite with ReplSuiteMixin { assertContains("res1: Int = 55", output) } -}
\ No newline at end of file + test("external vars") { + val output = runInterpreter(sparkUrl, """ + var v = 7 + sc.parallelize(1 to 10).map(x => v).take(10).reduceLeft(_+_) + v = 10 + sc.parallelize(1 to 10).map(x => v).take(10).reduceLeft(_+_) + """) + assertDoesNotContain("error:", output) + assertDoesNotContain("Exception", output) + assertContains("res0: Int = 70", output) + assertContains("res1: Int = 100", output) + } + + test("external classes") { + val output = runInterpreter(sparkUrl, """ + class C { + def foo = 5 + } + sc.parallelize(1 to 10).map(x => (new C).foo).take(10).reduceLeft(_+_) + """) + assertDoesNotContain("error:", output) + assertDoesNotContain("Exception", output) + assertContains("res0: Int = 50", output) + } + + test("external functions") { + val output = runInterpreter(sparkUrl, """ + def double(x: Int) = x + x + sc.parallelize(1 to 10).map(x => double(x)).take(10).reduceLeft(_+_) + """) + assertDoesNotContain("error:", output) + assertDoesNotContain("Exception", output) + assertContains("res0: Int = 110", output) + } + + test("external functions that access vars") { + val output = runInterpreter(sparkUrl, """ + var v = 7 + def getV() = v + sc.parallelize(1 to 10).map(x => getV()).take(10).reduceLeft(_+_) + v = 10 + sc.parallelize(1 to 10).map(x => getV()).take(10).reduceLeft(_+_) + """) + assertDoesNotContain("error:", output) + assertDoesNotContain("Exception", output) + assertContains("res0: Int = 70", output) + assertContains("res1: Int = 100", output) + } + + test("broadcast vars") { + // Test that the value that a broadcast var had when it was created is used, + // even if that variable is then modified in the driver program + + val output = runInterpreter(sparkUrl, """ + var array = new Array[Int](5) + val broadcastArray = sc.broadcast(array) + sc.parallelize(0 to 4).map(x => broadcastArray.value(x)).take(5) + array(0) = 5 + sc.parallelize(0 to 4).map(x => broadcastArray.value(x)).take(5) + """) + assertDoesNotContain("error:", output) + assertDoesNotContain("Exception", output) + assertContains("res0: Array[Int] = Array(0, 0, 0, 0, 0)", output) + assertContains("res2: Array[Int] = Array(5, 0, 0, 0, 0)", output) + } + + +} |