spark git commit: [CORE][TESTS] minor fix of JavaSerializerSuite
Repository: spark Updated Branches: refs/heads/master 0370abdfd -> 40e52a27c [CORE][TESTS] minor fix of JavaSerializerSuite Not jira is created. The original test is passed because the class cast is lazy (only when the object's method is invoked). Author: Jeff Zhang Closes #10371 from zjffdu/minor_fix. Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/40e52a27 Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/40e52a27 Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/40e52a27 Branch: refs/heads/master Commit: 40e52a27c74259237dd1906c0e8b54d2ae645dfb Parents: 0370abd Author: Jeff Zhang Authored: Fri Dec 18 00:49:56 2015 -0800 Committer: Reynold Xin Committed: Fri Dec 18 00:49:56 2015 -0800 -- .../org/apache/spark/serializer/JavaSerializerSuite.scala | 9 +++-- 1 file changed, 7 insertions(+), 2 deletions(-) -- http://git-wip-us.apache.org/repos/asf/spark/blob/40e52a27/core/src/test/scala/org/apache/spark/serializer/JavaSerializerSuite.scala -- diff --git a/core/src/test/scala/org/apache/spark/serializer/JavaSerializerSuite.scala b/core/src/test/scala/org/apache/spark/serializer/JavaSerializerSuite.scala index 20f4567..6a6ea42 100644 --- a/core/src/test/scala/org/apache/spark/serializer/JavaSerializerSuite.scala +++ b/core/src/test/scala/org/apache/spark/serializer/JavaSerializerSuite.scala @@ -23,13 +23,18 @@ class JavaSerializerSuite extends SparkFunSuite { test("JavaSerializer instances are serializable") { val serializer = new JavaSerializer(new SparkConf()) val instance = serializer.newInstance() -instance.deserialize[JavaSerializer](instance.serialize(serializer)) +val obj = instance.deserialize[JavaSerializer](instance.serialize(serializer)) +// enforce class cast +obj.getClass } test("Deserialize object containing a primitive Class as attribute") { val serializer = new JavaSerializer(new SparkConf()) val instance = serializer.newInstance() -instance.deserialize[JavaSerializer](instance.serialize(new ContainsPrimitiveClass())) +val obj = instance.deserialize[ContainsPrimitiveClass](instance.serialize( + new ContainsPrimitiveClass())) +// enforce class cast +obj.getClass } } - To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org
spark git commit: [SPARK-12413] Fix Mesos ZK persistence
Repository: spark Updated Branches: refs/heads/master 40e52a27c -> 2bebaa39d [SPARK-12413] Fix Mesos ZK persistence I believe this fixes SPARK-12413. I'm currently running an integration test to verify. Author: Michael Gummelt Closes #10366 from mgummelt/fix-zk-mesos. Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/2bebaa39 Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/2bebaa39 Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/2bebaa39 Branch: refs/heads/master Commit: 2bebaa39d9da33bc93ef682959cd42c1968a6a3e Parents: 40e52a2 Author: Michael Gummelt Authored: Fri Dec 18 20:18:00 2015 +0900 Committer: Kousuke Saruta Committed: Fri Dec 18 20:18:00 2015 +0900 -- .../org/apache/spark/deploy/rest/mesos/MesosRestServer.scala | 6 +- 1 file changed, 5 insertions(+), 1 deletion(-) -- http://git-wip-us.apache.org/repos/asf/spark/blob/2bebaa39/core/src/main/scala/org/apache/spark/deploy/rest/mesos/MesosRestServer.scala -- diff --git a/core/src/main/scala/org/apache/spark/deploy/rest/mesos/MesosRestServer.scala b/core/src/main/scala/org/apache/spark/deploy/rest/mesos/MesosRestServer.scala index c0b9359..87d0fa8 100644 --- a/core/src/main/scala/org/apache/spark/deploy/rest/mesos/MesosRestServer.scala +++ b/core/src/main/scala/org/apache/spark/deploy/rest/mesos/MesosRestServer.scala @@ -99,7 +99,11 @@ private[mesos] class MesosSubmitRequestServlet( // cause spark-submit script to look for files in SPARK_HOME instead. // We only need the ability to specify where to find spark-submit script // which user can user spark.executor.home or spark.home configurations. -val environmentVariables = request.environmentVariables.filterKeys(!_.equals("SPARK_HOME")) +// +// Do not use `filterKeys` here to avoid SI-6654, which breaks ZK persistence +val environmentVariables = request.environmentVariables.filter { case (k, _) => + k != "SPARK_HOME" +} val name = request.sparkProperties.get("spark.app.name").getOrElse(mainClass) // Construct driver description - To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org
spark git commit: [SPARK-12413] Fix Mesos ZK persistence
Repository: spark Updated Branches: refs/heads/branch-1.6 9177ea383 -> df0231952 [SPARK-12413] Fix Mesos ZK persistence I believe this fixes SPARK-12413. I'm currently running an integration test to verify. Author: Michael Gummelt Closes #10366 from mgummelt/fix-zk-mesos. (cherry picked from commit 2bebaa39d9da33bc93ef682959cd42c1968a6a3e) Signed-off-by: Kousuke Saruta Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/df023195 Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/df023195 Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/df023195 Branch: refs/heads/branch-1.6 Commit: df0231952e5542e9870f8dde9ecbd7ad9a50f847 Parents: 9177ea3 Author: Michael Gummelt Authored: Fri Dec 18 20:18:00 2015 +0900 Committer: Kousuke Saruta Committed: Fri Dec 18 20:21:42 2015 +0900 -- .../org/apache/spark/deploy/rest/mesos/MesosRestServer.scala | 6 +- 1 file changed, 5 insertions(+), 1 deletion(-) -- http://git-wip-us.apache.org/repos/asf/spark/blob/df023195/core/src/main/scala/org/apache/spark/deploy/rest/mesos/MesosRestServer.scala -- diff --git a/core/src/main/scala/org/apache/spark/deploy/rest/mesos/MesosRestServer.scala b/core/src/main/scala/org/apache/spark/deploy/rest/mesos/MesosRestServer.scala index 196338f..ec44b78 100644 --- a/core/src/main/scala/org/apache/spark/deploy/rest/mesos/MesosRestServer.scala +++ b/core/src/main/scala/org/apache/spark/deploy/rest/mesos/MesosRestServer.scala @@ -99,7 +99,11 @@ private[mesos] class MesosSubmitRequestServlet( // look for files in SPARK_HOME instead. We only need the ability to specify where to find // spark-submit script which user can user spark.executor.home or spark.home configurations // (SPARK-12345). -val environmentVariables = request.environmentVariables.filterKeys(!_.equals("SPARK_HOME")) + +// Do not use `filterKeys` here to avoid SI-6654, which breaks ZK persistence +val environmentVariables = request.environmentVariables.filter { case (k, _) => + k != "SPARK_HOME" +} val name = request.sparkProperties.get("spark.app.name").getOrElse(mainClass) // Construct driver description - To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org
spark git commit: [SPARK-10500][SPARKR] sparkr.zip cannot be created if /R/lib is unwritable
Repository: spark Updated Branches: refs/heads/branch-1.5 a8d14cc06 -> d2f71c27c [SPARK-10500][SPARKR] sparkr.zip cannot be created if /R/lib is unwritable Backport https://github.com/apache/spark/pull/9390 and https://github.com/apache/spark/pull/9744 to branch-1.5. Author: Sun Rui Author: Shivaram Venkataraman Closes #10372 from sun-rui/SPARK-10500-branch-1.5. Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/d2f71c27 Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/d2f71c27 Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/d2f71c27 Branch: refs/heads/branch-1.5 Commit: d2f71c27c0f43475a69791693f1684e106e9b475 Parents: a8d14cc Author: Sun Rui Authored: Fri Dec 18 07:14:37 2015 -0800 Committer: Shivaram Venkataraman Committed: Fri Dec 18 07:14:37 2015 -0800 -- R/install-dev.bat | 6 +++ R/install-dev.sh| 4 ++ R/pkg/R/sparkR.R| 14 ++- R/pkg/inst/profile/general.R| 3 +- R/pkg/inst/worker/daemon.R | 5 ++- R/pkg/inst/worker/worker.R | 3 +- .../scala/org/apache/spark/api/r/RBackend.scala | 1 + .../scala/org/apache/spark/api/r/RRDD.scala | 4 +- .../scala/org/apache/spark/api/r/RUtils.scala | 37 + .../org/apache/spark/deploy/RPackageUtils.scala | 26 .../scala/org/apache/spark/deploy/RRunner.scala | 5 ++- .../org/apache/spark/deploy/SparkSubmit.scala | 43 .../apache/spark/deploy/SparkSubmitSuite.scala | 9 ++-- make-distribution.sh| 1 + 14 files changed, 124 insertions(+), 37 deletions(-) -- http://git-wip-us.apache.org/repos/asf/spark/blob/d2f71c27/R/install-dev.bat -- diff --git a/R/install-dev.bat b/R/install-dev.bat index 008a5c6..ed1c91a 100644 --- a/R/install-dev.bat +++ b/R/install-dev.bat @@ -25,3 +25,9 @@ set SPARK_HOME=%~dp0.. MKDIR %SPARK_HOME%\R\lib R.exe CMD INSTALL --library="%SPARK_HOME%\R\lib" %SPARK_HOME%\R\pkg\ + +rem Zip the SparkR package so that it can be distributed to worker nodes on YARN +pushd %SPARK_HOME%\R\lib +%JAVA_HOME%\bin\jar.exe cfM "%SPARK_HOME%\R\lib\sparkr.zip" SparkR +popd + http://git-wip-us.apache.org/repos/asf/spark/blob/d2f71c27/R/install-dev.sh -- diff --git a/R/install-dev.sh b/R/install-dev.sh index 59d98c9..4972bb9 100755 --- a/R/install-dev.sh +++ b/R/install-dev.sh @@ -42,4 +42,8 @@ Rscript -e ' if("devtools" %in% rownames(installed.packages())) { library(devtoo # Install SparkR to $LIB_DIR R CMD INSTALL --library=$LIB_DIR $FWDIR/pkg/ +# Zip the SparkR package so that it can be distributed to worker nodes on YARN +cd $LIB_DIR +jar cfM "$LIB_DIR/sparkr.zip" SparkR + popd > /dev/null http://git-wip-us.apache.org/repos/asf/spark/blob/d2f71c27/R/pkg/R/sparkR.R -- diff --git a/R/pkg/R/sparkR.R b/R/pkg/R/sparkR.R index 26f941e..154fbb4 100644 --- a/R/pkg/R/sparkR.R +++ b/R/pkg/R/sparkR.R @@ -49,6 +49,12 @@ sparkR.stop <- function() { } } +# Remove the R package lib path from .libPaths() +if (exists(".libPath", envir = env)) { + libPath <- get(".libPath", envir = env) + .libPaths(.libPaths()[.libPaths() != libPath]) +} + if (exists(".backendLaunched", envir = env)) { callJStatic("SparkRHandler", "stopBackend") } @@ -149,14 +155,20 @@ sparkR.init <- function( f <- file(path, open="rb") backendPort <- readInt(f) monitorPort <- readInt(f) +rLibPath <- readString(f) close(f) file.remove(path) if (length(backendPort) == 0 || backendPort == 0 || -length(monitorPort) == 0 || monitorPort == 0) { +length(monitorPort) == 0 || monitorPort == 0 || +length(rLibPath) != 1) { stop("JVM failed to launch") } assign(".monitorConn", socketConnection(port = monitorPort), envir = .sparkREnv) assign(".backendLaunched", 1, envir = .sparkREnv) +if (rLibPath != "") { + assign(".libPath", rLibPath, envir = .sparkREnv) + .libPaths(c(rLibPath, .libPaths())) +} } .sparkREnv$backendPort <- backendPort http://git-wip-us.apache.org/repos/asf/spark/blob/d2f71c27/R/pkg/inst/profile/general.R -- diff --git a/R/pkg/inst/profile/general.R b/R/pkg/inst/profile/general.R index 2a8a821..c55fe9b 100644 --- a/R/pkg/inst/profile/general.R +++ b/R/pkg/inst/profile/general.R @@ -17,6 +17,7 @@ .First <- function() { packageDir <- Sys.getenv("SPARKR_PACKAGE_DIR") - .libPaths(c(p
spark git commit: [SPARK-12350][CORE] Don't log errors when requested stream is not found.
Repository: spark Updated Branches: refs/heads/master ea59b0f3a -> 278281828 [SPARK-12350][CORE] Don't log errors when requested stream is not found. If a client requests a non-existent stream, just send a failure message back, without logging any error on the server side (since it's not a server error). On the executor side, avoid error logs by translating any errors during transfer to a `ClassNotFoundException`, so that loading the class is retried on a the parent class loader. This can mask IO errors during transmission, but the most common cause is that the class is not served by the remote end. Author: Marcelo Vanzin Closes #10337 from vanzin/SPARK-12350. Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/27828182 Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/27828182 Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/27828182 Branch: refs/heads/master Commit: 2782818287a71925523c1320291db6cb25221e9f Parents: ea59b0f Author: Marcelo Vanzin Authored: Fri Dec 18 09:49:08 2015 -0800 Committer: Marcelo Vanzin Committed: Fri Dec 18 09:49:08 2015 -0800 -- .../apache/spark/rpc/netty/NettyRpcEnv.scala| 17 .../spark/rpc/netty/NettyStreamManager.scala| 7 +-- .../spark/network/server/StreamManager.java | 1 + .../network/server/TransportRequestHandler.java | 7 ++- .../apache/spark/repl/ExecutorClassLoader.scala | 21 ++-- 5 files changed, 39 insertions(+), 14 deletions(-) -- http://git-wip-us.apache.org/repos/asf/spark/blob/27828182/core/src/main/scala/org/apache/spark/rpc/netty/NettyRpcEnv.scala -- diff --git a/core/src/main/scala/org/apache/spark/rpc/netty/NettyRpcEnv.scala b/core/src/main/scala/org/apache/spark/rpc/netty/NettyRpcEnv.scala index de3db6b..975ea1a 100644 --- a/core/src/main/scala/org/apache/spark/rpc/netty/NettyRpcEnv.scala +++ b/core/src/main/scala/org/apache/spark/rpc/netty/NettyRpcEnv.scala @@ -363,15 +363,14 @@ private[netty] class NettyRpcEnv( } override def read(dst: ByteBuffer): Int = { - val result = if (error == null) { -Try(source.read(dst)) - } else { -Failure(error) - } - - result match { + Try(source.read(dst)) match { case Success(bytesRead) => bytesRead -case Failure(error) => throw error +case Failure(readErr) => + if (error != null) { +throw error + } else { +throw readErr + } } } @@ -397,7 +396,7 @@ private[netty] class NettyRpcEnv( } override def onFailure(streamId: String, cause: Throwable): Unit = { - logError(s"Error downloading stream $streamId.", cause) + logDebug(s"Error downloading stream $streamId.", cause) source.setError(cause) sink.close() } http://git-wip-us.apache.org/repos/asf/spark/blob/27828182/core/src/main/scala/org/apache/spark/rpc/netty/NettyStreamManager.scala -- diff --git a/core/src/main/scala/org/apache/spark/rpc/netty/NettyStreamManager.scala b/core/src/main/scala/org/apache/spark/rpc/netty/NettyStreamManager.scala index 394cde4..afcb023 100644 --- a/core/src/main/scala/org/apache/spark/rpc/netty/NettyStreamManager.scala +++ b/core/src/main/scala/org/apache/spark/rpc/netty/NettyStreamManager.scala @@ -58,8 +58,11 @@ private[netty] class NettyStreamManager(rpcEnv: NettyRpcEnv) new File(dir, fname) } -require(file != null && file.isFile(), s"File not found: $streamId") -new FileSegmentManagedBuffer(rpcEnv.transportConf, file, 0, file.length()) +if (file != null && file.isFile()) { + new FileSegmentManagedBuffer(rpcEnv.transportConf, file, 0, file.length()) +} else { + null +} } override def addFile(file: File): String = { http://git-wip-us.apache.org/repos/asf/spark/blob/27828182/network/common/src/main/java/org/apache/spark/network/server/StreamManager.java -- diff --git a/network/common/src/main/java/org/apache/spark/network/server/StreamManager.java b/network/common/src/main/java/org/apache/spark/network/server/StreamManager.java index 3f01559..07f161a 100644 --- a/network/common/src/main/java/org/apache/spark/network/server/StreamManager.java +++ b/network/common/src/main/java/org/apache/spark/network/server/StreamManager.java @@ -54,6 +54,7 @@ public abstract class StreamManager { * {@link #getChunk(long, int)} method. * * @param streamId id of a stream that has been previously registered with the StreamManager. + * @return A managed buffer for the stream, or null if the stream was not found. */
spark git commit: [SPARK-11619][SQL] cannot use UDTF in DataFrame.selectExpr
Repository: spark Updated Branches: refs/heads/master 278281828 -> ee444fe4b [SPARK-11619][SQL] cannot use UDTF in DataFrame.selectExpr Description of the problem from cloud-fan Actually this line: https://github.com/apache/spark/blob/branch-1.5/sql/core/src/main/scala/org/apache/spark/sql/DataFrame.scala#L689 When we use `selectExpr`, we pass in `UnresolvedFunction` to `DataFrame.select` and fall in the last case. A workaround is to do special handling for UDTF like we did for `explode`(and `json_tuple` in 1.6), wrap it with `MultiAlias`. Another workaround is using `expr`, for example, `df.select(expr("explode(a)").as(Nil))`, I think `selectExpr` is no longer needed after we have the `expr` function Author: Dilip Biswal Closes #9981 from dilipbiswal/spark-11619. Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/ee444fe4 Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/ee444fe4 Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/ee444fe4 Branch: refs/heads/master Commit: ee444fe4b8c9f382524e1fa346c67ba6da8104d8 Parents: 2782818 Author: Dilip Biswal Authored: Fri Dec 18 09:54:30 2015 -0800 Committer: Yin Huai Committed: Fri Dec 18 09:54:30 2015 -0800 -- .../apache/spark/sql/catalyst/analysis/Analyzer.scala | 12 ++-- .../apache/spark/sql/catalyst/analysis/unresolved.scala | 6 +- .../src/main/scala/org/apache/spark/sql/Column.scala| 12 +++- .../src/main/scala/org/apache/spark/sql/Dataset.scala | 2 +- .../scala/org/apache/spark/sql/DataFrameSuite.scala | 7 +++ .../scala/org/apache/spark/sql/JsonFunctionsSuite.scala | 4 .../main/scala/org/apache/spark/sql/hive/HiveQl.scala | 2 +- 7 files changed, 31 insertions(+), 14 deletions(-) -- http://git-wip-us.apache.org/repos/asf/spark/blob/ee444fe4/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala -- diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala index 64dd83a..c396546 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala @@ -149,12 +149,12 @@ class Analyzer( exprs.zipWithIndex.map { case (expr, i) => expr transform { -case u @ UnresolvedAlias(child) => child match { +case u @ UnresolvedAlias(child, optionalAliasName) => child match { case ne: NamedExpression => ne case e if !e.resolved => u case g: Generator => MultiAlias(g, Nil) case c @ Cast(ne: NamedExpression, _) => Alias(c, ne.name)() - case other => Alias(other, s"_c$i")() + case other => Alias(other, optionalAliasName.getOrElse(s"_c$i"))() } } }.asInstanceOf[Seq[NamedExpression]] @@ -287,7 +287,7 @@ class Analyzer( } } val newGroupByExprs = groupByExprs.map { - case UnresolvedAlias(e) => e + case UnresolvedAlias(e, _) => e case e => e } Aggregate(newGroupByExprs, groupByExprs ++ pivotAggregates, child) @@ -352,19 +352,19 @@ class Analyzer( Project( projectList.flatMap { case s: Star => s.expand(child, resolver) -case UnresolvedAlias(f @ UnresolvedFunction(_, args, _)) if containsStar(args) => +case UnresolvedAlias(f @ UnresolvedFunction(_, args, _), _) if containsStar(args) => val newChildren = expandStarExpressions(args, child) UnresolvedAlias(child = f.copy(children = newChildren)) :: Nil case Alias(f @ UnresolvedFunction(_, args, _), name) if containsStar(args) => val newChildren = expandStarExpressions(args, child) Alias(child = f.copy(children = newChildren), name)() :: Nil -case UnresolvedAlias(c @ CreateArray(args)) if containsStar(args) => +case UnresolvedAlias(c @ CreateArray(args), _) if containsStar(args) => val expandedArgs = args.flatMap { case s: Star => s.expand(child, resolver) case o => o :: Nil } UnresolvedAlias(c.copy(children = expandedArgs)) :: Nil -case UnresolvedAlias(c @ CreateStruct(args)) if containsStar(args) => +case UnresolvedAlias(c @ CreateStruct(args), _) if containsStar(args) => val expandedArgs = args.flatMap { case s: Star => s.expand(child, resolver) case o => o :: Nil http:/
spark git commit: [SPARK-12054] [SQL] Consider nullability of expression in codegen
Repository: spark Updated Branches: refs/heads/master ee444fe4b -> 4af647c77 [SPARK-12054] [SQL] Consider nullability of expression in codegen This could simplify the generated code for expressions that is not nullable. This PR fix lots of bugs about nullability. Author: Davies Liu Closes #10333 from davies/skip_nullable. Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/4af647c7 Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/4af647c7 Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/4af647c7 Branch: refs/heads/master Commit: 4af647c77ded6a0d3087ceafb2e30e01d97e7a06 Parents: ee444fe Author: Davies Liu Authored: Fri Dec 18 10:09:17 2015 -0800 Committer: Davies Liu Committed: Fri Dec 18 10:09:17 2015 -0800 -- .../catalyst/expressions/BoundAttribute.scala | 17 ++-- .../spark/sql/catalyst/expressions/Cast.scala | 28 +++--- .../sql/catalyst/expressions/Expression.scala | 95 +--- .../aggregate/CentralMomentAgg.scala| 2 +- .../catalyst/expressions/aggregate/Corr.scala | 3 +- .../catalyst/expressions/aggregate/Count.scala | 19 ++-- .../catalyst/expressions/aggregate/Sum.scala| 24 +++-- .../expressions/codegen/CodegenFallback.scala | 27 -- .../codegen/GenerateMutableProjection.scala | 65 +- .../codegen/GenerateUnsafeProjection.scala | 21 +++-- .../expressions/complexTypeExtractors.scala | 19 ++-- .../expressions/datetimeExpressions.scala | 4 + .../expressions/decimalExpressions.scala| 1 + .../catalyst/expressions/jsonExpressions.scala | 15 ++-- .../catalyst/expressions/mathExpressions.scala | 5 ++ .../spark/sql/catalyst/expressions/misc.scala | 1 + .../expressions/stringExpressions.scala | 1 + .../expressions/windowExpressions.scala | 4 +- .../catalyst/plans/logical/basicOperators.scala | 9 +- .../sql/catalyst/expressions/CastSuite.scala| 10 +-- .../catalyst/expressions/ComplexTypeSuite.scala | 1 - .../scala/org/apache/spark/sql/DataFrame.scala | 21 +++-- .../org/apache/spark/sql/execution/Window.scala | 9 +- .../apache/spark/sql/execution/commands.scala | 2 +- .../spark/sql/execution/datasources/ddl.scala | 2 +- .../sql/execution/joins/HashOuterJoin.scala | 80 + .../execution/joins/SortMergeOuterJoin.scala| 2 +- 27 files changed, 261 insertions(+), 226 deletions(-) -- http://git-wip-us.apache.org/repos/asf/spark/blob/4af647c7/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/BoundAttribute.scala -- diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/BoundAttribute.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/BoundAttribute.scala index ff1f28d..7293d5d 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/BoundAttribute.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/BoundAttribute.scala @@ -69,10 +69,17 @@ case class BoundReference(ordinal: Int, dataType: DataType, nullable: Boolean) override def genCode(ctx: CodeGenContext, ev: GeneratedExpressionCode): String = { val javaType = ctx.javaType(dataType) val value = ctx.getValue(ctx.INPUT_ROW, dataType, ordinal.toString) -s""" - boolean ${ev.isNull} = ${ctx.INPUT_ROW}.isNullAt($ordinal); - $javaType ${ev.value} = ${ev.isNull} ? ${ctx.defaultValue(dataType)} : ($value); -""" +if (nullable) { + s""" +boolean ${ev.isNull} = ${ctx.INPUT_ROW}.isNullAt($ordinal); +$javaType ${ev.value} = ${ev.isNull} ? ${ctx.defaultValue(dataType)} : ($value); + """ +} else { + ev.isNull = "false" + s""" +$javaType ${ev.value} = $value; + """ +} } } @@ -92,7 +99,7 @@ object BindReferences extends Logging { sys.error(s"Couldn't find $a in ${input.mkString("[", ",", "]")}") } } else { - BoundReference(ordinal, a.dataType, a.nullable) + BoundReference(ordinal, a.dataType, input(ordinal).nullable) } } }.asInstanceOf[A] // Kind of a hack, but safe. TODO: Tighten return type when possible. http://git-wip-us.apache.org/repos/asf/spark/blob/4af647c7/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Cast.scala -- diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Cast.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Cast.scala index cb60d59..b18f49f 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Cast.scala +++ b
spark git commit: [SPARK-12218][SQL] Invalid splitting of nested AND expressions in Data Source filter API
Repository: spark Updated Branches: refs/heads/master 4af647c77 -> 41ee7c57a [SPARK-12218][SQL] Invalid splitting of nested AND expressions in Data Source filter API JIRA: https://issues.apache.org/jira/browse/SPARK-12218 When creating filters for Parquet/ORC, we should not push nested AND expressions partially. Author: Yin Huai Closes #10362 from yhuai/SPARK-12218. Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/41ee7c57 Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/41ee7c57 Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/41ee7c57 Branch: refs/heads/master Commit: 41ee7c57abd9f52065fd7ffb71a8af229603371d Parents: 4af647c Author: Yin Huai Authored: Fri Dec 18 10:52:14 2015 -0800 Committer: Yin Huai Committed: Fri Dec 18 10:53:13 2015 -0800 -- .../datasources/parquet/ParquetFilters.scala| 12 ++- .../parquet/ParquetFilterSuite.scala| 19 + .../apache/spark/sql/hive/orc/OrcFilters.scala | 22 +--- .../sql/hive/orc/OrcHadoopFsRelationSuite.scala | 20 ++ 4 files changed, 60 insertions(+), 13 deletions(-) -- http://git-wip-us.apache.org/repos/asf/spark/blob/41ee7c57/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala -- diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala index 0771432..883013b 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala @@ -257,7 +257,17 @@ private[sql] object ParquetFilters { makeGtEq.lift(dataTypeOf(name)).map(_(name, value)) case sources.And(lhs, rhs) => -(createFilter(schema, lhs) ++ createFilter(schema, rhs)).reduceOption(FilterApi.and) +// At here, it is not safe to just convert one side if we do not understand the +// other side. Here is an example used to explain the reason. +// Let's say we have NOT(a = 2 AND b in ('1')) and we do not understand how to +// convert b in ('1'). If we only convert a = 2, we will end up with a filter +// NOT(a = 2), which will generate wrong results. +// Pushing one side of AND down is only safe to do at the top level. +// You can see ParquetRelation's initializeLocalJobFunc method as an example. +for { + lhsFilter <- createFilter(schema, lhs) + rhsFilter <- createFilter(schema, rhs) +} yield FilterApi.and(lhsFilter, rhsFilter) case sources.Or(lhs, rhs) => for { http://git-wip-us.apache.org/repos/asf/spark/blob/41ee7c57/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala -- diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala index 6178e37..045425f 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala @@ -362,4 +362,23 @@ class ParquetFilterSuite extends QueryTest with ParquetTest with SharedSQLContex } } } + + test("SPARK-12218: 'Not' is included in Parquet filter pushdown") { +import testImplicits._ + +withSQLConf(SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED.key -> "true") { + withTempPath { dir => +val path = s"${dir.getCanonicalPath}/table1" +(1 to 5).map(i => (i, (i % 2).toString)).toDF("a", "b").write.parquet(path) + +checkAnswer( + sqlContext.read.parquet(path).where("not (a = 2) or not(b in ('1'))"), + (1 to 5).map(i => Row(i, (i % 2).toString))) + +checkAnswer( + sqlContext.read.parquet(path).where("not (a = 2 and b in ('1'))"), + (1 to 5).map(i => Row(i, (i % 2).toString))) + } +} + } } http://git-wip-us.apache.org/repos/asf/spark/blob/41ee7c57/sql/hive/src/main/scala/org/apache/spark/sql/hive/orc/OrcFilters.scala -- diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/orc/OrcFilters.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/orc/OrcFilters.scala index 27193f5..ebfb175 100644 --- a/sql/hive/src/main/scala/org/apache/spark/sql/hi
spark git commit: [SPARK-12218][SQL] Invalid splitting of nested AND expressions in Data Source filter API
Repository: spark Updated Branches: refs/heads/branch-1.6 df0231952 -> 1dc71ec77 [SPARK-12218][SQL] Invalid splitting of nested AND expressions in Data Source filter API JIRA: https://issues.apache.org/jira/browse/SPARK-12218 When creating filters for Parquet/ORC, we should not push nested AND expressions partially. Author: Yin Huai Closes #10362 from yhuai/SPARK-12218. (cherry picked from commit 41ee7c57abd9f52065fd7ffb71a8af229603371d) Signed-off-by: Yin Huai Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/1dc71ec7 Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/1dc71ec7 Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/1dc71ec7 Branch: refs/heads/branch-1.6 Commit: 1dc71ec777ff7cac5d3d7adb13f2d63ffe8909b6 Parents: df02319 Author: Yin Huai Authored: Fri Dec 18 10:52:14 2015 -0800 Committer: Yin Huai Committed: Fri Dec 18 10:53:31 2015 -0800 -- .../datasources/parquet/ParquetFilters.scala| 12 ++- .../parquet/ParquetFilterSuite.scala| 19 + .../apache/spark/sql/hive/orc/OrcFilters.scala | 22 +--- .../sql/hive/orc/OrcHadoopFsRelationSuite.scala | 20 ++ 4 files changed, 60 insertions(+), 13 deletions(-) -- http://git-wip-us.apache.org/repos/asf/spark/blob/1dc71ec7/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala -- diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala index 0771432..883013b 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala @@ -257,7 +257,17 @@ private[sql] object ParquetFilters { makeGtEq.lift(dataTypeOf(name)).map(_(name, value)) case sources.And(lhs, rhs) => -(createFilter(schema, lhs) ++ createFilter(schema, rhs)).reduceOption(FilterApi.and) +// At here, it is not safe to just convert one side if we do not understand the +// other side. Here is an example used to explain the reason. +// Let's say we have NOT(a = 2 AND b in ('1')) and we do not understand how to +// convert b in ('1'). If we only convert a = 2, we will end up with a filter +// NOT(a = 2), which will generate wrong results. +// Pushing one side of AND down is only safe to do at the top level. +// You can see ParquetRelation's initializeLocalJobFunc method as an example. +for { + lhsFilter <- createFilter(schema, lhs) + rhsFilter <- createFilter(schema, rhs) +} yield FilterApi.and(lhsFilter, rhsFilter) case sources.Or(lhs, rhs) => for { http://git-wip-us.apache.org/repos/asf/spark/blob/1dc71ec7/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala -- diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala index cc5aae0..ed5a352 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala @@ -364,4 +364,23 @@ class ParquetFilterSuite extends QueryTest with ParquetTest with SharedSQLContex } } } + + test("SPARK-12218: 'Not' is included in Parquet filter pushdown") { +import testImplicits._ + +withSQLConf(SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED.key -> "true") { + withTempPath { dir => +val path = s"${dir.getCanonicalPath}/table1" +(1 to 5).map(i => (i, (i % 2).toString)).toDF("a", "b").write.parquet(path) + +checkAnswer( + sqlContext.read.parquet(path).where("not (a = 2) or not(b in ('1'))"), + (1 to 5).map(i => Row(i, (i % 2).toString))) + +checkAnswer( + sqlContext.read.parquet(path).where("not (a = 2 and b in ('1'))"), + (1 to 5).map(i => Row(i, (i % 2).toString))) + } +} + } } http://git-wip-us.apache.org/repos/asf/spark/blob/1dc71ec7/sql/hive/src/main/scala/org/apache/spark/sql/hive/orc/OrcFilters.scala -- diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/orc/OrcFilters.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/o
spark git commit: [SPARK-12218][SQL] Invalid splitting of nested AND expressions in Data Source filter API
Repository: spark Updated Branches: refs/heads/branch-1.5 d2f71c27c -> afffe24c0 [SPARK-12218][SQL] Invalid splitting of nested AND expressions in Data Source filter API JIRA: https://issues.apache.org/jira/browse/SPARK-12218 When creating filters for Parquet/ORC, we should not push nested AND expressions partially. Author: Yin Huai Closes #10362 from yhuai/SPARK-12218. (cherry picked from commit 41ee7c57abd9f52065fd7ffb71a8af229603371d) Signed-off-by: Yin Huai Conflicts: sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/afffe24c Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/afffe24c Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/afffe24c Branch: refs/heads/branch-1.5 Commit: afffe24c00f49c199383bd07530266a8124d77f2 Parents: d2f71c2 Author: Yin Huai Authored: Fri Dec 18 10:52:14 2015 -0800 Committer: Yin Huai Committed: Fri Dec 18 10:55:21 2015 -0800 -- .../datasources/parquet/ParquetFilters.scala| 12 ++- .../parquet/ParquetFilterSuite.scala| 19 + .../apache/spark/sql/hive/orc/OrcFilters.scala | 22 +--- .../sql/hive/orc/OrcHadoopFsRelationSuite.scala | 20 ++ 4 files changed, 60 insertions(+), 13 deletions(-) -- http://git-wip-us.apache.org/repos/asf/spark/blob/afffe24c/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala -- diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala index 1f0405f..3f7a409 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala @@ -264,7 +264,17 @@ private[sql] object ParquetFilters { makeGtEq.lift(dataTypeOf(name)).map(_(name, value)) case sources.And(lhs, rhs) => -(createFilter(schema, lhs) ++ createFilter(schema, rhs)).reduceOption(FilterApi.and) +// At here, it is not safe to just convert one side if we do not understand the +// other side. Here is an example used to explain the reason. +// Let's say we have NOT(a = 2 AND b in ('1')) and we do not understand how to +// convert b in ('1'). If we only convert a = 2, we will end up with a filter +// NOT(a = 2), which will generate wrong results. +// Pushing one side of AND down is only safe to do at the top level. +// You can see ParquetRelation's initializeLocalJobFunc method as an example. +for { + lhsFilter <- createFilter(schema, lhs) + rhsFilter <- createFilter(schema, rhs) +} yield FilterApi.and(lhsFilter, rhsFilter) case sources.Or(lhs, rhs) => for { http://git-wip-us.apache.org/repos/asf/spark/blob/afffe24c/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala -- diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala index f88ddc7..05d305a 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala @@ -336,4 +336,23 @@ class ParquetFilterSuite extends QueryTest with ParquetTest with SharedSQLContex } } } + + test("SPARK-12218: 'Not' is included in Parquet filter pushdown") { +import testImplicits._ + +withSQLConf(SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED.key -> "true") { + withTempPath { dir => +val path = s"${dir.getCanonicalPath}/table1" +(1 to 5).map(i => (i, (i % 2).toString)).toDF("a", "b").write.parquet(path) + +checkAnswer( + sqlContext.read.parquet(path).where("not (a = 2) or not(b in ('1'))"), + (1 to 5).map(i => Row(i, (i % 2).toString))) + +checkAnswer( + sqlContext.read.parquet(path).where("not (a = 2 and b in ('1'))"), + (1 to 5).map(i => Row(i, (i % 2).toString))) + } +} + } } http://git-wip-us.apache.org/repos/asf/spark/blob/afffe24c/sql/hive/src/main/scala/org/apache/spark/sql/hive/orc/OrcFilters.scala -- diff --git a/sql/
spark git commit: Revert "[SPARK-12365][CORE] Use ShutdownHookManager where Runtime.getRuntime.addShutdownHook() is called"
Repository: spark Updated Branches: refs/heads/branch-1.6 1dc71ec77 -> 3b903e44b Revert "[SPARK-12365][CORE] Use ShutdownHookManager where Runtime.getRuntime.addShutdownHook() is called" This reverts commit 4af64385b085002d94c54d11bbd144f9f026bbd8. Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/3b903e44 Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/3b903e44 Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/3b903e44 Branch: refs/heads/branch-1.6 Commit: 3b903e44b912cd36ec26e9e95444656eee7b0c46 Parents: 1dc71ec Author: Andrew Or Authored: Fri Dec 18 12:56:03 2015 -0800 Committer: Andrew Or Committed: Fri Dec 18 12:56:03 2015 -0800 -- .../spark/deploy/ExternalShuffleService.scala | 18 -- .../deploy/mesos/MesosClusterDispatcher.scala | 13 - .../apache/spark/util/ShutdownHookManager.scala | 4 scalastyle-config.xml | 12 4 files changed, 20 insertions(+), 27 deletions(-) -- http://git-wip-us.apache.org/repos/asf/spark/blob/3b903e44/core/src/main/scala/org/apache/spark/deploy/ExternalShuffleService.scala -- diff --git a/core/src/main/scala/org/apache/spark/deploy/ExternalShuffleService.scala b/core/src/main/scala/org/apache/spark/deploy/ExternalShuffleService.scala index 7fc96e4..e8a1e35 100644 --- a/core/src/main/scala/org/apache/spark/deploy/ExternalShuffleService.scala +++ b/core/src/main/scala/org/apache/spark/deploy/ExternalShuffleService.scala @@ -28,7 +28,7 @@ import org.apache.spark.network.sasl.SaslServerBootstrap import org.apache.spark.network.server.{TransportServerBootstrap, TransportServer} import org.apache.spark.network.shuffle.ExternalShuffleBlockHandler import org.apache.spark.network.util.TransportConf -import org.apache.spark.util.{ShutdownHookManager, Utils} +import org.apache.spark.util.Utils /** * Provides a server from which Executors can read shuffle files (rather than reading directly from @@ -118,13 +118,19 @@ object ExternalShuffleService extends Logging { server = newShuffleService(sparkConf, securityManager) server.start() -ShutdownHookManager.addShutdownHook { () => - logInfo("Shutting down shuffle service.") - server.stop() - barrier.countDown() -} +installShutdownHook() // keep running until the process is terminated barrier.await() } + + private def installShutdownHook(): Unit = { +Runtime.getRuntime.addShutdownHook(new Thread("External Shuffle Service shutdown thread") { + override def run() { +logInfo("Shutting down shuffle service.") +server.stop() +barrier.countDown() + } +}) + } } http://git-wip-us.apache.org/repos/asf/spark/blob/3b903e44/core/src/main/scala/org/apache/spark/deploy/mesos/MesosClusterDispatcher.scala -- diff --git a/core/src/main/scala/org/apache/spark/deploy/mesos/MesosClusterDispatcher.scala b/core/src/main/scala/org/apache/spark/deploy/mesos/MesosClusterDispatcher.scala index 389eff5..5d4e5b8 100644 --- a/core/src/main/scala/org/apache/spark/deploy/mesos/MesosClusterDispatcher.scala +++ b/core/src/main/scala/org/apache/spark/deploy/mesos/MesosClusterDispatcher.scala @@ -22,7 +22,7 @@ import java.util.concurrent.CountDownLatch import org.apache.spark.deploy.mesos.ui.MesosClusterUI import org.apache.spark.deploy.rest.mesos.MesosRestServer import org.apache.spark.scheduler.cluster.mesos._ -import org.apache.spark.util.{ShutdownHookManager, SignalLogger} +import org.apache.spark.util.SignalLogger import org.apache.spark.{Logging, SecurityManager, SparkConf} /* @@ -103,11 +103,14 @@ private[mesos] object MesosClusterDispatcher extends Logging { } val dispatcher = new MesosClusterDispatcher(dispatcherArgs, conf) dispatcher.start() -ShutdownHookManager.addShutdownHook { () => - logInfo("Shutdown hook is shutting down dispatcher") - dispatcher.stop() - dispatcher.awaitShutdown() +val shutdownHook = new Thread() { + override def run() { +logInfo("Shutdown hook is shutting down dispatcher") +dispatcher.stop() +dispatcher.awaitShutdown() + } } +Runtime.getRuntime.addShutdownHook(shutdownHook) dispatcher.awaitShutdown() } } http://git-wip-us.apache.org/repos/asf/spark/blob/3b903e44/core/src/main/scala/org/apache/spark/util/ShutdownHookManager.scala -- diff --git a/core/src/main/scala/org/apache/spark/util/ShutdownHookManager.scala b/core/src/main/scala/org/apache/spark/util/ShutdownHookManager.scala index 1a0f3b4..620f226 100644 ---
spark git commit: [SPARK-12404][SQL] Ensure objects passed to StaticInvoke is Serializable
Repository: spark Updated Branches: refs/heads/master 41ee7c57a -> 6eba65525 [SPARK-12404][SQL] Ensure objects passed to StaticInvoke is Serializable Now `StaticInvoke` receives `Any` as a object and `StaticInvoke` can be serialized but sometimes the object passed is not serializable. For example, following code raises Exception because `RowEncoder#extractorsFor` invoked indirectly makes `StaticInvoke`. ``` case class TimestampContainer(timestamp: java.sql.Timestamp) val rdd = sc.parallelize(1 to 2).map(_ => TimestampContainer(System.currentTimeMillis)) val df = rdd.toDF val ds = df.as[TimestampContainer] val rdd2 = ds.rdd <- invokes extractorsFor indirectory ``` I'll add test cases. Author: Kousuke Saruta Author: Michael Armbrust Closes #10357 from sarutak/SPARK-12404. Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/6eba6552 Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/6eba6552 Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/6eba6552 Branch: refs/heads/master Commit: 6eba655259d2bcea27d0147b37d5d1e476e85422 Parents: 41ee7c5 Author: Kousuke Saruta Authored: Fri Dec 18 14:05:06 2015 -0800 Committer: Michael Armbrust Committed: Fri Dec 18 14:05:06 2015 -0800 -- .../spark/sql/catalyst/JavaTypeInference.scala | 12 ++--- .../spark/sql/catalyst/ScalaReflection.scala| 16 +++--- .../sql/catalyst/encoders/RowEncoder.scala | 14 +++--- .../sql/catalyst/expressions/objects.scala | 8 ++- .../org/apache/spark/sql/JavaDatasetSuite.java | 52 .../org/apache/spark/sql/DatasetSuite.scala | 12 + 6 files changed, 88 insertions(+), 26 deletions(-) -- http://git-wip-us.apache.org/repos/asf/spark/blob/6eba6552/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala -- diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala index c8ee87e..f566d1b 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala @@ -194,7 +194,7 @@ object JavaTypeInference { case c if c == classOf[java.sql.Date] => StaticInvoke( - DateTimeUtils, + DateTimeUtils.getClass, ObjectType(c), "toJavaDate", getPath :: Nil, @@ -202,7 +202,7 @@ object JavaTypeInference { case c if c == classOf[java.sql.Timestamp] => StaticInvoke( - DateTimeUtils, + DateTimeUtils.getClass, ObjectType(c), "toJavaTimestamp", getPath :: Nil, @@ -276,7 +276,7 @@ object JavaTypeInference { ObjectType(classOf[Array[Any]])) StaticInvoke( - ArrayBasedMapData, + ArrayBasedMapData.getClass, ObjectType(classOf[JMap[_, _]]), "toJavaMap", keyData :: valueData :: Nil) @@ -341,21 +341,21 @@ object JavaTypeInference { case c if c == classOf[java.sql.Timestamp] => StaticInvoke( -DateTimeUtils, +DateTimeUtils.getClass, TimestampType, "fromJavaTimestamp", inputObject :: Nil) case c if c == classOf[java.sql.Date] => StaticInvoke( -DateTimeUtils, +DateTimeUtils.getClass, DateType, "fromJavaDate", inputObject :: Nil) case c if c == classOf[java.math.BigDecimal] => StaticInvoke( -Decimal, +Decimal.getClass, DecimalType.SYSTEM_DEFAULT, "apply", inputObject :: Nil) http://git-wip-us.apache.org/repos/asf/spark/blob/6eba6552/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/ScalaReflection.scala -- diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/ScalaReflection.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/ScalaReflection.scala index ecff860..c1b1d5c 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/ScalaReflection.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/ScalaReflection.scala @@ -223,7 +223,7 @@ object ScalaReflection extends ScalaReflection { case t if t <:< localTypeOf[java.sql.Date] => StaticInvoke( - DateTimeUtils, + DateTimeUtils.getClass, ObjectType(classOf[java.sql.Date]), "toJavaDate", getPath :: Nil, @@ -231,7 +231,7
spark git commit: [SPARK-12404][SQL] Ensure objects passed to StaticInvoke is Serializable
Repository: spark Updated Branches: refs/heads/branch-1.6 3b903e44b -> bd33d4ee8 [SPARK-12404][SQL] Ensure objects passed to StaticInvoke is Serializable Now `StaticInvoke` receives `Any` as a object and `StaticInvoke` can be serialized but sometimes the object passed is not serializable. For example, following code raises Exception because `RowEncoder#extractorsFor` invoked indirectly makes `StaticInvoke`. ``` case class TimestampContainer(timestamp: java.sql.Timestamp) val rdd = sc.parallelize(1 to 2).map(_ => TimestampContainer(System.currentTimeMillis)) val df = rdd.toDF val ds = df.as[TimestampContainer] val rdd2 = ds.rdd <- invokes extractorsFor indirectory ``` I'll add test cases. Author: Kousuke Saruta Author: Michael Armbrust Closes #10357 from sarutak/SPARK-12404. (cherry picked from commit 6eba655259d2bcea27d0147b37d5d1e476e85422) Signed-off-by: Michael Armbrust Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/bd33d4ee Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/bd33d4ee Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/bd33d4ee Branch: refs/heads/branch-1.6 Commit: bd33d4ee847973289a58032df35375f03e9f9865 Parents: 3b903e4 Author: Kousuke Saruta Authored: Fri Dec 18 14:05:06 2015 -0800 Committer: Michael Armbrust Committed: Fri Dec 18 14:05:16 2015 -0800 -- .../spark/sql/catalyst/JavaTypeInference.scala | 12 ++--- .../spark/sql/catalyst/ScalaReflection.scala| 16 +++--- .../sql/catalyst/encoders/RowEncoder.scala | 14 +++--- .../sql/catalyst/expressions/objects.scala | 8 ++- .../org/apache/spark/sql/JavaDatasetSuite.java | 52 .../org/apache/spark/sql/DatasetSuite.scala | 12 + 6 files changed, 88 insertions(+), 26 deletions(-) -- http://git-wip-us.apache.org/repos/asf/spark/blob/bd33d4ee/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala -- diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala index c8ee87e..f566d1b 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala @@ -194,7 +194,7 @@ object JavaTypeInference { case c if c == classOf[java.sql.Date] => StaticInvoke( - DateTimeUtils, + DateTimeUtils.getClass, ObjectType(c), "toJavaDate", getPath :: Nil, @@ -202,7 +202,7 @@ object JavaTypeInference { case c if c == classOf[java.sql.Timestamp] => StaticInvoke( - DateTimeUtils, + DateTimeUtils.getClass, ObjectType(c), "toJavaTimestamp", getPath :: Nil, @@ -276,7 +276,7 @@ object JavaTypeInference { ObjectType(classOf[Array[Any]])) StaticInvoke( - ArrayBasedMapData, + ArrayBasedMapData.getClass, ObjectType(classOf[JMap[_, _]]), "toJavaMap", keyData :: valueData :: Nil) @@ -341,21 +341,21 @@ object JavaTypeInference { case c if c == classOf[java.sql.Timestamp] => StaticInvoke( -DateTimeUtils, +DateTimeUtils.getClass, TimestampType, "fromJavaTimestamp", inputObject :: Nil) case c if c == classOf[java.sql.Date] => StaticInvoke( -DateTimeUtils, +DateTimeUtils.getClass, DateType, "fromJavaDate", inputObject :: Nil) case c if c == classOf[java.math.BigDecimal] => StaticInvoke( -Decimal, +Decimal.getClass, DecimalType.SYSTEM_DEFAULT, "apply", inputObject :: Nil) http://git-wip-us.apache.org/repos/asf/spark/blob/bd33d4ee/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/ScalaReflection.scala -- diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/ScalaReflection.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/ScalaReflection.scala index 9b6b5b8..ea98956 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/ScalaReflection.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/ScalaReflection.scala @@ -223,7 +223,7 @@ object ScalaReflection extends ScalaReflection { case t if t <:< localTypeOf[java.sql.Date] => StaticInvoke( - DateTimeUtils, + DateTimeUtils.getClass,
spark git commit: [SPARK-11985][STREAMING][KINESIS][DOCS] Update Kinesis docs
Repository: spark Updated Branches: refs/heads/master 6eba65525 -> 2377b707f [SPARK-11985][STREAMING][KINESIS][DOCS] Update Kinesis docs - Provide example on `message handler` - Provide bit on KPL record de-aggregation - Fix typos Author: Burak Yavuz Closes #9970 from brkyvz/kinesis-docs. Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/2377b707 Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/2377b707 Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/2377b707 Branch: refs/heads/master Commit: 2377b707f25449f4557bf048bb384c743d9008e5 Parents: 6eba655 Author: Burak Yavuz Authored: Fri Dec 18 15:24:41 2015 -0800 Committer: Shixiong Zhu Committed: Fri Dec 18 15:24:41 2015 -0800 -- docs/streaming-kinesis-integration.md | 54 +- 1 file changed, 45 insertions(+), 9 deletions(-) -- http://git-wip-us.apache.org/repos/asf/spark/blob/2377b707/docs/streaming-kinesis-integration.md -- diff --git a/docs/streaming-kinesis-integration.md b/docs/streaming-kinesis-integration.md index 238a911..07194b0 100644 --- a/docs/streaming-kinesis-integration.md +++ b/docs/streaming-kinesis-integration.md @@ -23,7 +23,7 @@ A Kinesis stream can be set up at one of the valid Kinesis endpoints with 1 or m **Note that by linking to this library, you will include [ASL](https://aws.amazon.com/asl/)-licensed code in your application.** -2. **Programming:** In the streaming application code, import `KinesisUtils` and create the input DStream as follows: +2. **Programming:** In the streaming application code, import `KinesisUtils` and create the input DStream of byte array as follows: @@ -36,7 +36,7 @@ A Kinesis stream can be set up at one of the valid Kinesis endpoints with 1 or m [region name], [initial position], [checkpoint interval], StorageLevel.MEMORY_AND_DISK_2) See the [API docs](api/scala/index.html#org.apache.spark.streaming.kinesis.KinesisUtils$) - and the [example]({{site.SPARK_GITHUB_URL}}/tree/master/extras/kinesis-asl/src/main/scala/org/apache/spark/examples/streaming/KinesisWordCountASL.scala). Refer to the Running the Example section for instructions on how to run the example. + and the [example]({{site.SPARK_GITHUB_URL}}/tree/master/extras/kinesis-asl/src/main/scala/org/apache/spark/examples/streaming/KinesisWordCountASL.scala). Refer to the [Running the Example](#running-the-example) subsection for instructions on how to run the example. @@ -49,7 +49,7 @@ A Kinesis stream can be set up at one of the valid Kinesis endpoints with 1 or m [region name], [initial position], [checkpoint interval], StorageLevel.MEMORY_AND_DISK_2); See the [API docs](api/java/index.html?org/apache/spark/streaming/kinesis/KinesisUtils.html) - and the [example]({{site.SPARK_GITHUB_URL}}/tree/master/extras/kinesis-asl/src/main/java/org/apache/spark/examples/streaming/JavaKinesisWordCountASL.java). Refer to the next subsection for instructions to run the example. + and the [example]({{site.SPARK_GITHUB_URL}}/tree/master/extras/kinesis-asl/src/main/java/org/apache/spark/examples/streaming/JavaKinesisWordCountASL.java). Refer to the [Running the Example](#running-the-example) subsection for instructions to run the example. @@ -60,18 +60,47 @@ A Kinesis stream can be set up at one of the valid Kinesis endpoints with 1 or m [region name], [initial position], [checkpoint interval], StorageLevel.MEMORY_AND_DISK_2) See the [API docs](api/python/pyspark.streaming.html#pyspark.streaming.kinesis.KinesisUtils) - and the [example]({{site.SPARK_GITHUB_URL}}/tree/master/extras/kinesis-asl/src/main/python/examples/streaming/kinesis_wordcount_asl.py). Refer to the next subsection for instructions to run the example. + and the [example]({{site.SPARK_GITHUB_URL}}/tree/master/extras/kinesis-asl/src/main/python/examples/streaming/kinesis_wordcount_asl.py). Refer to the [Running the Example](#running-the-example) subsection for instructions to run the example. -- `streamingContext`: StreamingContext containg an application name used by Kinesis to tie this Kinesis application to the Kinesis stream + You may also provide a "message handler function" that takes a Kinesis `Record` and returns a generic object `T`, in case you would like to use other data included in a `Record` such as partition key. This is currently only supported in Scala and Java. - - `[Kineiss app name]`: The application name that will be used to checkpoint the Kinesis + + + +
spark git commit: [SPARK-11985][STREAMING][KINESIS][DOCS] Update Kinesis docs
Repository: spark Updated Branches: refs/heads/branch-1.6 bd33d4ee8 -> eca401ee5 [SPARK-11985][STREAMING][KINESIS][DOCS] Update Kinesis docs - Provide example on `message handler` - Provide bit on KPL record de-aggregation - Fix typos Author: Burak Yavuz Closes #9970 from brkyvz/kinesis-docs. (cherry picked from commit 2377b707f25449f4557bf048bb384c743d9008e5) Signed-off-by: Shixiong Zhu Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/eca401ee Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/eca401ee Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/eca401ee Branch: refs/heads/branch-1.6 Commit: eca401ee5d3ae683cbee531c1f8bc981f9603fc8 Parents: bd33d4e Author: Burak Yavuz Authored: Fri Dec 18 15:24:41 2015 -0800 Committer: Shixiong Zhu Committed: Fri Dec 18 15:24:49 2015 -0800 -- docs/streaming-kinesis-integration.md | 54 +- 1 file changed, 45 insertions(+), 9 deletions(-) -- http://git-wip-us.apache.org/repos/asf/spark/blob/eca401ee/docs/streaming-kinesis-integration.md -- diff --git a/docs/streaming-kinesis-integration.md b/docs/streaming-kinesis-integration.md index 238a911..07194b0 100644 --- a/docs/streaming-kinesis-integration.md +++ b/docs/streaming-kinesis-integration.md @@ -23,7 +23,7 @@ A Kinesis stream can be set up at one of the valid Kinesis endpoints with 1 or m **Note that by linking to this library, you will include [ASL](https://aws.amazon.com/asl/)-licensed code in your application.** -2. **Programming:** In the streaming application code, import `KinesisUtils` and create the input DStream as follows: +2. **Programming:** In the streaming application code, import `KinesisUtils` and create the input DStream of byte array as follows: @@ -36,7 +36,7 @@ A Kinesis stream can be set up at one of the valid Kinesis endpoints with 1 or m [region name], [initial position], [checkpoint interval], StorageLevel.MEMORY_AND_DISK_2) See the [API docs](api/scala/index.html#org.apache.spark.streaming.kinesis.KinesisUtils$) - and the [example]({{site.SPARK_GITHUB_URL}}/tree/master/extras/kinesis-asl/src/main/scala/org/apache/spark/examples/streaming/KinesisWordCountASL.scala). Refer to the Running the Example section for instructions on how to run the example. + and the [example]({{site.SPARK_GITHUB_URL}}/tree/master/extras/kinesis-asl/src/main/scala/org/apache/spark/examples/streaming/KinesisWordCountASL.scala). Refer to the [Running the Example](#running-the-example) subsection for instructions on how to run the example. @@ -49,7 +49,7 @@ A Kinesis stream can be set up at one of the valid Kinesis endpoints with 1 or m [region name], [initial position], [checkpoint interval], StorageLevel.MEMORY_AND_DISK_2); See the [API docs](api/java/index.html?org/apache/spark/streaming/kinesis/KinesisUtils.html) - and the [example]({{site.SPARK_GITHUB_URL}}/tree/master/extras/kinesis-asl/src/main/java/org/apache/spark/examples/streaming/JavaKinesisWordCountASL.java). Refer to the next subsection for instructions to run the example. + and the [example]({{site.SPARK_GITHUB_URL}}/tree/master/extras/kinesis-asl/src/main/java/org/apache/spark/examples/streaming/JavaKinesisWordCountASL.java). Refer to the [Running the Example](#running-the-example) subsection for instructions to run the example. @@ -60,18 +60,47 @@ A Kinesis stream can be set up at one of the valid Kinesis endpoints with 1 or m [region name], [initial position], [checkpoint interval], StorageLevel.MEMORY_AND_DISK_2) See the [API docs](api/python/pyspark.streaming.html#pyspark.streaming.kinesis.KinesisUtils) - and the [example]({{site.SPARK_GITHUB_URL}}/tree/master/extras/kinesis-asl/src/main/python/examples/streaming/kinesis_wordcount_asl.py). Refer to the next subsection for instructions to run the example. + and the [example]({{site.SPARK_GITHUB_URL}}/tree/master/extras/kinesis-asl/src/main/python/examples/streaming/kinesis_wordcount_asl.py). Refer to the [Running the Example](#running-the-example) subsection for instructions to run the example. -- `streamingContext`: StreamingContext containg an application name used by Kinesis to tie this Kinesis application to the Kinesis stream + You may also provide a "message handler function" that takes a Kinesis `Record` and returns a generic object `T`, in case you would like to use other data included in a `Record` such as partition key. This is currently only supported in Scala and Java. - - `[Kineiss
spark git commit: [SPARK-12411][CORE] Decrease executor heartbeat timeout to match heartbeat interval
Repository: spark Updated Branches: refs/heads/master 60da0e11f -> 0514e8d4b [SPARK-12411][CORE] Decrease executor heartbeat timeout to match heartbeat interval Previously, the rpc timeout was the default network timeout, which is the same value the driver uses to determine dead executors. This means if there is a network issue, the executor is determined dead after one heartbeat attempt. There is a separate config for the heartbeat interval which is a better value to use for the heartbeat RPC. With this change, the executor will make multiple heartbeat attempts even with RPC issues. Author: Nong Li Closes #10365 from nongli/spark-12411. Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/0514e8d4 Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/0514e8d4 Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/0514e8d4 Branch: refs/heads/master Commit: 0514e8d4b69615ba8918649e7e3c46b5713b6540 Parents: 60da0e1 Author: Nong Li Authored: Fri Dec 18 16:05:18 2015 -0800 Committer: Andrew Or Committed: Fri Dec 18 16:05:18 2015 -0800 -- core/src/main/scala/org/apache/spark/executor/Executor.scala | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) -- http://git-wip-us.apache.org/repos/asf/spark/blob/0514e8d4/core/src/main/scala/org/apache/spark/executor/Executor.scala -- diff --git a/core/src/main/scala/org/apache/spark/executor/Executor.scala b/core/src/main/scala/org/apache/spark/executor/Executor.scala index 552b644..9b14184 100644 --- a/core/src/main/scala/org/apache/spark/executor/Executor.scala +++ b/core/src/main/scala/org/apache/spark/executor/Executor.scala @@ -30,6 +30,7 @@ import scala.util.control.NonFatal import org.apache.spark._ import org.apache.spark.deploy.SparkHadoopUtil import org.apache.spark.memory.TaskMemoryManager +import org.apache.spark.rpc.RpcTimeout import org.apache.spark.scheduler.{DirectTaskResult, IndirectTaskResult, Task} import org.apache.spark.shuffle.FetchFailedException import org.apache.spark.storage.{StorageLevel, TaskResultBlockId} @@ -445,7 +446,8 @@ private[spark] class Executor( val message = Heartbeat(executorId, tasksMetrics.toArray, env.blockManager.blockManagerId) try { - val response = heartbeatReceiverRef.askWithRetry[HeartbeatResponse](message) + val response = heartbeatReceiverRef.askWithRetry[HeartbeatResponse]( + message, RpcTimeout(conf, "spark.executor.heartbeatInterval", "10s")) if (response.reregisterBlockManager) { logInfo("Told to re-register on heartbeat") env.blockManager.reregister() - To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org
spark git commit: [SPARK-11097][CORE] Add channelActive callback to RpcHandler to monitor the new connections
Repository: spark Updated Branches: refs/heads/master 0514e8d4b -> 007a32f90 [SPARK-11097][CORE] Add channelActive callback to RpcHandler to monitor the new connections Added `channelActive` to `RpcHandler` so that `NettyRpcHandler` doesn't need `clients` any more. Author: Shixiong Zhu Closes #10301 from zsxwing/network-events. Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/007a32f9 Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/007a32f9 Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/007a32f9 Branch: refs/heads/master Commit: 007a32f90af1065bfa3ca4cdb194c40c06e87abf Parents: 0514e8d Author: Shixiong Zhu Authored: Fri Dec 18 16:06:37 2015 -0800 Committer: Andrew Or Committed: Fri Dec 18 16:06:37 2015 -0800 -- .../mesos/MesosExternalShuffleService.scala | 2 +- .../apache/spark/rpc/netty/NettyRpcEnv.scala| 17 +-- .../org/apache/spark/rpc/RpcEnvSuite.scala | 148 +++ .../spark/rpc/netty/NettyRpcHandlerSuite.scala | 6 +- .../client/TransportResponseHandler.java| 6 +- .../spark/network/sasl/SaslRpcHandler.java | 9 +- .../spark/network/server/MessageHandler.java| 7 +- .../apache/spark/network/server/RpcHandler.java | 9 +- .../network/server/TransportChannelHandler.java | 21 ++- .../network/server/TransportRequestHandler.java | 9 +- .../spark/network/sasl/SparkSaslSuite.java | 6 +- 11 files changed, 148 insertions(+), 92 deletions(-) -- http://git-wip-us.apache.org/repos/asf/spark/blob/007a32f9/core/src/main/scala/org/apache/spark/deploy/mesos/MesosExternalShuffleService.scala -- diff --git a/core/src/main/scala/org/apache/spark/deploy/mesos/MesosExternalShuffleService.scala b/core/src/main/scala/org/apache/spark/deploy/mesos/MesosExternalShuffleService.scala index 8ffcfc0..4172d92 100644 --- a/core/src/main/scala/org/apache/spark/deploy/mesos/MesosExternalShuffleService.scala +++ b/core/src/main/scala/org/apache/spark/deploy/mesos/MesosExternalShuffleService.scala @@ -65,7 +65,7 @@ private[mesos] class MesosExternalShuffleBlockHandler(transportConf: TransportCo /** * On connection termination, clean up shuffle files written by the associated application. */ - override def connectionTerminated(client: TransportClient): Unit = { + override def channelInactive(client: TransportClient): Unit = { val address = client.getSocketAddress if (connectedApps.contains(address)) { val appId = connectedApps(address) http://git-wip-us.apache.org/repos/asf/spark/blob/007a32f9/core/src/main/scala/org/apache/spark/rpc/netty/NettyRpcEnv.scala -- diff --git a/core/src/main/scala/org/apache/spark/rpc/netty/NettyRpcEnv.scala b/core/src/main/scala/org/apache/spark/rpc/netty/NettyRpcEnv.scala index 975ea1a..090a1b9 100644 --- a/core/src/main/scala/org/apache/spark/rpc/netty/NettyRpcEnv.scala +++ b/core/src/main/scala/org/apache/spark/rpc/netty/NettyRpcEnv.scala @@ -548,10 +548,6 @@ private[netty] class NettyRpcHandler( nettyEnv: NettyRpcEnv, streamManager: StreamManager) extends RpcHandler with Logging { - // TODO: Can we add connection callback (channel registered) to the underlying framework? - // A variable to track whether we should dispatch the RemoteProcessConnected message. - private val clients = new ConcurrentHashMap[TransportClient, JBoolean]() - // A variable to track the remote RpcEnv addresses of all clients private val remoteAddresses = new ConcurrentHashMap[RpcAddress, RpcAddress]() @@ -574,9 +570,6 @@ private[netty] class NettyRpcHandler( val addr = client.getChannel().remoteAddress().asInstanceOf[InetSocketAddress] assert(addr != null) val clientAddr = RpcAddress(addr.getHostName, addr.getPort) -if (clients.putIfAbsent(client, JBoolean.TRUE) == null) { - dispatcher.postToAll(RemoteProcessConnected(clientAddr)) -} val requestMessage = nettyEnv.deserialize[RequestMessage](client, message) if (requestMessage.senderAddress == null) { // Create a new message with the socket address of the client as the sender. @@ -613,10 +606,16 @@ private[netty] class NettyRpcHandler( } } - override def connectionTerminated(client: TransportClient): Unit = { + override def channelActive(client: TransportClient): Unit = { +val addr = client.getChannel().remoteAddress().asInstanceOf[InetSocketAddress] +assert(addr != null) +val clientAddr = RpcAddress(addr.getHostName, addr.getPort) +dispatcher.postToAll(RemoteProcessConnected(clientAddr)) + } + + override def channelInactive(client: TransportClient): Unit = { val addr = client.getChannel.remoteA
spark git commit: [SPARK-12345][CORE] Do not send SPARK_HOME through Spark submit REST interface
Repository: spark Updated Branches: refs/heads/master 007a32f90 -> ba9332edd [SPARK-12345][CORE] Do not send SPARK_HOME through Spark submit REST interface It is usually an invalid location on the remote machine executing the job. It is picked up by the Mesos support in cluster mode, and most of the time causes the job to fail. Fixes SPARK-12345 Author: Luc Bourlier Closes #10329 from skyluc/issue/SPARK_HOME. Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/ba9332ed Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/ba9332ed Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/ba9332ed Branch: refs/heads/master Commit: ba9332edd889730c906404041bc83b1643d80961 Parents: 007a32f Author: Luc Bourlier Authored: Fri Dec 18 16:21:01 2015 -0800 Committer: Andrew Or Committed: Fri Dec 18 16:21:01 2015 -0800 -- .../org/apache/spark/deploy/rest/RestSubmissionClient.scala| 6 -- 1 file changed, 4 insertions(+), 2 deletions(-) -- http://git-wip-us.apache.org/repos/asf/spark/blob/ba9332ed/core/src/main/scala/org/apache/spark/deploy/rest/RestSubmissionClient.scala -- diff --git a/core/src/main/scala/org/apache/spark/deploy/rest/RestSubmissionClient.scala b/core/src/main/scala/org/apache/spark/deploy/rest/RestSubmissionClient.scala index f0dd667..0744c64 100644 --- a/core/src/main/scala/org/apache/spark/deploy/rest/RestSubmissionClient.scala +++ b/core/src/main/scala/org/apache/spark/deploy/rest/RestSubmissionClient.scala @@ -428,8 +428,10 @@ private[spark] object RestSubmissionClient { * Filter non-spark environment variables from any environment. */ private[rest] def filterSystemEnvironment(env: Map[String, String]): Map[String, String] = { -env.filter { case (k, _) => - (k.startsWith("SPARK_") && k != "SPARK_ENV_LOADED") || k.startsWith("MESOS_") +env.filterKeys { k => + // SPARK_HOME is filtered out because it is usually wrong on the remote machine (SPARK-12345) + (k.startsWith("SPARK_") && k != "SPARK_ENV_LOADED" && k != "SPARK_HOME") || +k.startsWith("MESOS_") } } } - To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org
[3/3] spark git commit: Revert "[SPARK-12345][MESOS] Filter SPARK_HOME when submitting Spark jobs with Mesos cluster mode."
Revert "[SPARK-12345][MESOS] Filter SPARK_HOME when submitting Spark jobs with Mesos cluster mode." This reverts commit ad8c1f0b840284d05da737fb2cc5ebf8848f4490. Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/a78a91f4 Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/a78a91f4 Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/a78a91f4 Branch: refs/heads/master Commit: a78a91f4d7239c14bd5d0b18cdc87d55594a8d8a Parents: 8a9417b Author: Andrew Or Authored: Fri Dec 18 16:22:51 2015 -0800 Committer: Andrew Or Committed: Fri Dec 18 16:22:51 2015 -0800 -- .../org/apache/spark/deploy/rest/mesos/MesosRestServer.scala | 7 +-- .../spark/scheduler/cluster/mesos/MesosSchedulerUtils.scala | 2 +- 2 files changed, 2 insertions(+), 7 deletions(-) -- http://git-wip-us.apache.org/repos/asf/spark/blob/a78a91f4/core/src/main/scala/org/apache/spark/deploy/rest/mesos/MesosRestServer.scala -- diff --git a/core/src/main/scala/org/apache/spark/deploy/rest/mesos/MesosRestServer.scala b/core/src/main/scala/org/apache/spark/deploy/rest/mesos/MesosRestServer.scala index 24510db..868cc35 100644 --- a/core/src/main/scala/org/apache/spark/deploy/rest/mesos/MesosRestServer.scala +++ b/core/src/main/scala/org/apache/spark/deploy/rest/mesos/MesosRestServer.scala @@ -94,12 +94,7 @@ private[mesos] class MesosSubmitRequestServlet( val driverMemory = sparkProperties.get("spark.driver.memory") val driverCores = sparkProperties.get("spark.driver.cores") val appArgs = request.appArgs -// We don't want to pass down SPARK_HOME when launching Spark apps -// with Mesos cluster mode since it's populated by default on the client and it will -// cause spark-submit script to look for files in SPARK_HOME instead. -// We only need the ability to specify where to find spark-submit script -// which user can user spark.executor.home or spark.home configurations. -val environmentVariables = request.environmentVariables.filter(!_.equals("SPARK_HOME")) +val environmentVariables = request.environmentVariables val name = request.sparkProperties.get("spark.app.name").getOrElse(mainClass) // Construct driver description http://git-wip-us.apache.org/repos/asf/spark/blob/a78a91f4/core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerUtils.scala -- diff --git a/core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerUtils.scala b/core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerUtils.scala index 573355b..721861f 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerUtils.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerUtils.scala @@ -34,7 +34,7 @@ import org.apache.spark.util.Utils /** * Shared trait for implementing a Mesos Scheduler. This holds common state and helper - * methods the Mesos scheduler will use. + * methods and Mesos scheduler will use. */ private[mesos] trait MesosSchedulerUtils extends Logging { // Lock used to wait for scheduler to be registered - To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org
[1/3] spark git commit: Revert "[SPARK-12413] Fix Mesos ZK persistence"
Repository: spark Updated Branches: refs/heads/master ba9332edd -> a78a91f4d Revert "[SPARK-12413] Fix Mesos ZK persistence" This reverts commit 2bebaa39d9da33bc93ef682959cd42c1968a6a3e. Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/14be5dec Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/14be5dec Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/14be5dec Branch: refs/heads/master Commit: 14be5dece291c900bf97d87b850d906717645fd4 Parents: ba9332e Author: Andrew Or Authored: Fri Dec 18 16:22:33 2015 -0800 Committer: Andrew Or Committed: Fri Dec 18 16:22:33 2015 -0800 -- .../org/apache/spark/deploy/rest/mesos/MesosRestServer.scala | 6 +- 1 file changed, 1 insertion(+), 5 deletions(-) -- http://git-wip-us.apache.org/repos/asf/spark/blob/14be5dec/core/src/main/scala/org/apache/spark/deploy/rest/mesos/MesosRestServer.scala -- diff --git a/core/src/main/scala/org/apache/spark/deploy/rest/mesos/MesosRestServer.scala b/core/src/main/scala/org/apache/spark/deploy/rest/mesos/MesosRestServer.scala index 87d0fa8..c0b9359 100644 --- a/core/src/main/scala/org/apache/spark/deploy/rest/mesos/MesosRestServer.scala +++ b/core/src/main/scala/org/apache/spark/deploy/rest/mesos/MesosRestServer.scala @@ -99,11 +99,7 @@ private[mesos] class MesosSubmitRequestServlet( // cause spark-submit script to look for files in SPARK_HOME instead. // We only need the ability to specify where to find spark-submit script // which user can user spark.executor.home or spark.home configurations. -// -// Do not use `filterKeys` here to avoid SI-6654, which breaks ZK persistence -val environmentVariables = request.environmentVariables.filter { case (k, _) => - k != "SPARK_HOME" -} +val environmentVariables = request.environmentVariables.filterKeys(!_.equals("SPARK_HOME")) val name = request.sparkProperties.get("spark.app.name").getOrElse(mainClass) // Construct driver description - To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org
[2/3] spark git commit: Revert "[SPARK-12345][MESOS] Properly filter out SPARK_HOME in the Mesos REST server"
Revert "[SPARK-12345][MESOS] Properly filter out SPARK_HOME in the Mesos REST server" This reverts commit 8184568810e8a2e7d5371db2c6a0366ef4841f70. Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/8a9417bc Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/8a9417bc Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/8a9417bc Branch: refs/heads/master Commit: 8a9417bc4b227c57bdb0a5a38a225dbdf6e69f64 Parents: 14be5de Author: Andrew Or Authored: Fri Dec 18 16:22:41 2015 -0800 Committer: Andrew Or Committed: Fri Dec 18 16:22:41 2015 -0800 -- .../scala/org/apache/spark/deploy/rest/mesos/MesosRestServer.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) -- http://git-wip-us.apache.org/repos/asf/spark/blob/8a9417bc/core/src/main/scala/org/apache/spark/deploy/rest/mesos/MesosRestServer.scala -- diff --git a/core/src/main/scala/org/apache/spark/deploy/rest/mesos/MesosRestServer.scala b/core/src/main/scala/org/apache/spark/deploy/rest/mesos/MesosRestServer.scala index c0b9359..24510db 100644 --- a/core/src/main/scala/org/apache/spark/deploy/rest/mesos/MesosRestServer.scala +++ b/core/src/main/scala/org/apache/spark/deploy/rest/mesos/MesosRestServer.scala @@ -99,7 +99,7 @@ private[mesos] class MesosSubmitRequestServlet( // cause spark-submit script to look for files in SPARK_HOME instead. // We only need the ability to specify where to find spark-submit script // which user can user spark.executor.home or spark.home configurations. -val environmentVariables = request.environmentVariables.filterKeys(!_.equals("SPARK_HOME")) +val environmentVariables = request.environmentVariables.filter(!_.equals("SPARK_HOME")) val name = request.sparkProperties.get("spark.app.name").getOrElse(mainClass) // Construct driver description - To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org
spark git commit: [SPARK-12091] [PYSPARK] Deprecate the JAVA-specific deserialized storage levels
Repository: spark Updated Branches: refs/heads/master a78a91f4d -> 499ac3e69 [SPARK-12091] [PYSPARK] Deprecate the JAVA-specific deserialized storage levels The current default storage level of Python persist API is MEMORY_ONLY_SER. This is different from the default level MEMORY_ONLY in the official document and RDD APIs. davies Is this inconsistency intentional? Thanks! Updates: Since the data is always serialized on the Python side, the storage levels of JAVA-specific deserialization are not removed, such as MEMORY_ONLY. Updates: Based on the reviewers' feedback. In Python, stored objects will always be serialized with the [Pickle](https://docs.python.org/2/library/pickle.html) library, so it does not matter whether you choose a serialized level. The available storage levels in Python include `MEMORY_ONLY`, `MEMORY_ONLY_2`, `MEMORY_AND_DISK`, `MEMORY_AND_DISK_2`, `DISK_ONLY`, `DISK_ONLY_2` and `OFF_HEAP`. Author: gatorsmile Closes #10092 from gatorsmile/persistStorageLevel. Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/499ac3e6 Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/499ac3e6 Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/499ac3e6 Branch: refs/heads/master Commit: 499ac3e69a102f9b10a1d7e14382fa191516f7b5 Parents: a78a91f Author: gatorsmile Authored: Fri Dec 18 20:06:05 2015 -0800 Committer: Davies Liu Committed: Fri Dec 18 20:06:05 2015 -0800 -- docs/configuration.md | 7 --- docs/programming-guide.md | 10 ++ python/pyspark/rdd.py | 8 python/pyspark/sql/dataframe.py | 6 +++--- python/pyspark/storagelevel.py | 31 +-- python/pyspark/streaming/context.py | 2 +- python/pyspark/streaming/dstream.py | 4 ++-- python/pyspark/streaming/flume.py | 4 ++-- python/pyspark/streaming/kafka.py | 2 +- python/pyspark/streaming/mqtt.py| 2 +- 10 files changed, 45 insertions(+), 31 deletions(-) -- http://git-wip-us.apache.org/repos/asf/spark/blob/499ac3e6/docs/configuration.md -- diff --git a/docs/configuration.md b/docs/configuration.md index 38d3d05..85e7d12 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -687,9 +687,10 @@ Apart from these, the following properties are also available, and may be useful spark.rdd.compress false -Whether to compress serialized RDD partitions (e.g. for -StorageLevel.MEMORY_ONLY_SER). Can save substantial space at the cost of some -extra CPU time. +Whether to compress serialized RDD partitions (e.g. for +StorageLevel.MEMORY_ONLY_SER in Java +and Scala or StorageLevel.MEMORY_ONLY in Python). +Can save substantial space at the cost of some extra CPU time. http://git-wip-us.apache.org/repos/asf/spark/blob/499ac3e6/docs/programming-guide.md -- diff --git a/docs/programming-guide.md b/docs/programming-guide.md index f823b89..c5e2a1c 100644 --- a/docs/programming-guide.md +++ b/docs/programming-guide.md @@ -1196,14 +1196,14 @@ storage levels is: partitions that don't fit on disk, and read them from there when they're needed. - MEMORY_ONLY_SER + MEMORY_ONLY_SER (Java and Scala) Store RDD as serialized Java objects (one byte array per partition). This is generally more space-efficient than deserialized objects, especially when using a fast serializer, but more CPU-intensive to read. - MEMORY_AND_DISK_SER + MEMORY_AND_DISK_SER (Java and Scala) Similar to MEMORY_ONLY_SER, but spill partitions that don't fit in memory to disk instead of recomputing them on the fly each time they're needed. @@ -1230,7 +1230,9 @@ storage levels is: -**Note:** *In Python, stored objects will always be serialized with the [Pickle](https://docs.python.org/2/library/pickle.html) library, so it does not matter whether you choose a serialized level.* +**Note:** *In Python, stored objects will always be serialized with the [Pickle](https://docs.python.org/2/library/pickle.html) library, +so it does not matter whether you choose a serialized level. The available storage levels in Python include `MEMORY_ONLY`, `MEMORY_ONLY_2`, +`MEMORY_AND_DISK`, `MEMORY_AND_DISK_2`, `DISK_ONLY`, `DISK_ONLY_2` and `OFF_HEAP`.* Spark also automatically persists some intermediate data in shuffle operations (e.g. `reduceByKey`), even without users calling `persist`. This is done to avoid recomputing the entire input if a node fails during the shuffle. We still recommend users call `persist` on the resulting RDD if they plan to reuse it. @@ -1243,7 +1245,7 @@ efficiency. We recommend going t