spark git commit: [SPARK-11897][SQL] Add @scala.annotations.varargs to sql functions
Repository: spark
Updated Branches:
refs/heads/master 4021a28ac -> 12eea834d
[SPARK-11897][SQL] Add @scala.annotations.varargs to sql functions
Author: Xiu Guo
Closes #9918 from xguo27/SPARK-11897.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/12eea834
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/12eea834
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/12eea834
Branch: refs/heads/master
Commit: 12eea834d7382fbaa9c92182b682b8724049d7c1
Parents: 4021a28
Author: Xiu Guo
Authored: Tue Nov 24 00:07:40 2015 -0800
Committer: Reynold Xin
Committed: Tue Nov 24 00:07:40 2015 -0800
--
sql/core/src/main/scala/org/apache/spark/sql/functions.scala | 2 ++
1 file changed, 2 insertions(+)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/12eea834/sql/core/src/main/scala/org/apache/spark/sql/functions.scala
--
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/functions.scala
b/sql/core/src/main/scala/org/apache/spark/sql/functions.scala
index b27b134..6137ce3 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/functions.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/functions.scala
@@ -689,6 +689,7 @@ object functions extends LegacyFunctions {
* @group normal_funcs
* @since 1.4.0
*/
+ @scala.annotation.varargs
def array(colName: String, colNames: String*): Column = {
array((colName +: colNames).map(col) : _*)
}
@@ -871,6 +872,7 @@ object functions extends LegacyFunctions {
* @group normal_funcs
* @since 1.4.0
*/
+ @scala.annotation.varargs
def struct(colName: String, colNames: String*): Column = {
struct((colName +: colNames).map(col) : _*)
}
-
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
spark git commit: [SPARK-11897][SQL] Add @scala.annotations.varargs to sql functions
Repository: spark
Updated Branches:
refs/heads/branch-1.6 21e63929c -> e1b0a2376
[SPARK-11897][SQL] Add @scala.annotations.varargs to sql functions
Author: Xiu Guo
Closes #9918 from xguo27/SPARK-11897.
(cherry picked from commit 12eea834d7382fbaa9c92182b682b8724049d7c1)
Signed-off-by: Reynold Xin
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/e1b0a237
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/e1b0a237
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/e1b0a237
Branch: refs/heads/branch-1.6
Commit: e1b0a2376d9a78c14195bf50bd02004554097fba
Parents: 21e6392
Author: Xiu Guo
Authored: Tue Nov 24 00:07:40 2015 -0800
Committer: Reynold Xin
Committed: Tue Nov 24 00:07:48 2015 -0800
--
sql/core/src/main/scala/org/apache/spark/sql/functions.scala | 2 ++
1 file changed, 2 insertions(+)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/e1b0a237/sql/core/src/main/scala/org/apache/spark/sql/functions.scala
--
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/functions.scala
b/sql/core/src/main/scala/org/apache/spark/sql/functions.scala
index b27b134..6137ce3 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/functions.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/functions.scala
@@ -689,6 +689,7 @@ object functions extends LegacyFunctions {
* @group normal_funcs
* @since 1.4.0
*/
+ @scala.annotation.varargs
def array(colName: String, colNames: String*): Column = {
array((colName +: colNames).map(col) : _*)
}
@@ -871,6 +872,7 @@ object functions extends LegacyFunctions {
* @group normal_funcs
* @since 1.4.0
*/
+ @scala.annotation.varargs
def struct(colName: String, colNames: String*): Column = {
struct((colName +: colNames).map(col) : _*)
}
-
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
spark git commit: [SPARK-11906][WEB UI] Speculation Tasks Cause ProgressBar UI Overflow
Repository: spark
Updated Branches:
refs/heads/branch-1.6 e1b0a2376 -> 17ea95133
[SPARK-11906][WEB UI] Speculation Tasks Cause ProgressBar UI Overflow
When there are speculative tasks in the stage, running progress bar could
overflow and goes hidden on a new line:
3 completed / 2 running (including 1 speculative) out of 4 total tasks
This is a simple fix by capping the started tasks at `total - completed` tasks
I should note my preferred way to fix it is via css style
```css
.progress { display: flex; }
```
which shifts the correction burden from driver to web browser. However I
couldn't get selenium test to measure the position/dimension of the progress
bar correctly to get this unit tested.
It also has the side effect that the width will be calibrated so the running
occupies 2 / 5 instead of 1 / 4.
All in all, since this cosmetic bug is minor enough, I suppose the original
simple fix should be good enough.
Author: Forest Fang
Closes #9896 from saurfang/progressbar.
(cherry picked from commit 800bd799acf7f10a469d8d6537279953129eb2c6)
Signed-off-by: Sean Owen
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/17ea9513
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/17ea9513
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/17ea9513
Branch: refs/heads/branch-1.6
Commit: 17ea9513313187cfb6e7c7e66e6e469c5cef63cf
Parents: e1b0a23
Author: Forest Fang
Authored: Tue Nov 24 09:03:32 2015 +
Committer: Sean Owen
Committed: Tue Nov 24 09:03:44 2015 +
--
core/src/main/scala/org/apache/spark/ui/UIUtils.scala | 4 +++-
.../src/test/scala/org/apache/spark/ui/UIUtilsSuite.scala | 10 ++
2 files changed, 13 insertions(+), 1 deletion(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/17ea9513/core/src/main/scala/org/apache/spark/ui/UIUtils.scala
--
diff --git a/core/src/main/scala/org/apache/spark/ui/UIUtils.scala
b/core/src/main/scala/org/apache/spark/ui/UIUtils.scala
index 25dcb60..84a1116 100644
--- a/core/src/main/scala/org/apache/spark/ui/UIUtils.scala
+++ b/core/src/main/scala/org/apache/spark/ui/UIUtils.scala
@@ -319,7 +319,9 @@ private[spark] object UIUtils extends Logging {
skipped: Int,
total: Int): Seq[Node] = {
val completeWidth = "width: %s%%".format((completed.toDouble/total)*100)
-val startWidth = "width: %s%%".format((started.toDouble/total)*100)
+// started + completed can be > total when there are speculative tasks
+val boundedStarted = math.min(started, total - completed)
+val startWidth = "width: %s%%".format((boundedStarted.toDouble/total)*100)
http://git-wip-us.apache.org/repos/asf/spark/blob/17ea9513/core/src/test/scala/org/apache/spark/ui/UIUtilsSuite.scala
--
diff --git a/core/src/test/scala/org/apache/spark/ui/UIUtilsSuite.scala
b/core/src/test/scala/org/apache/spark/ui/UIUtilsSuite.scala
index 2b693c1..dd8d5ec 100644
--- a/core/src/test/scala/org/apache/spark/ui/UIUtilsSuite.scala
+++ b/core/src/test/scala/org/apache/spark/ui/UIUtilsSuite.scala
@@ -57,6 +57,16 @@ class UIUtilsSuite extends SparkFunSuite {
)
}
+ test("SPARK-11906: Progress bar should not overflow because of speculative
tasks") {
+val generated = makeProgressBar(2, 3, 0, 0, 4).head.child.filter(_.label
== "div")
+val expected = Seq(
+ ,
+
+)
+assert(generated.sameElements(expected),
+ s"\nRunning progress bar should round
down\n\nExpected:\n$expected\nGenerated:\n$generated")
+ }
+
private def verify(
desc: String, expected: Elem, errorMsg: String = "", baseUrl: String =
""): Unit = {
val generated = makeDescription(desc, baseUrl)
-
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
spark git commit: [SPARK-11906][WEB UI] Speculation Tasks Cause ProgressBar UI Overflow
Repository: spark
Updated Branches:
refs/heads/master 12eea834d -> 800bd799a
[SPARK-11906][WEB UI] Speculation Tasks Cause ProgressBar UI Overflow
When there are speculative tasks in the stage, running progress bar could
overflow and goes hidden on a new line:
3 completed / 2 running (including 1 speculative) out of 4 total tasks
This is a simple fix by capping the started tasks at `total - completed` tasks
I should note my preferred way to fix it is via css style
```css
.progress { display: flex; }
```
which shifts the correction burden from driver to web browser. However I
couldn't get selenium test to measure the position/dimension of the progress
bar correctly to get this unit tested.
It also has the side effect that the width will be calibrated so the running
occupies 2 / 5 instead of 1 / 4.
All in all, since this cosmetic bug is minor enough, I suppose the original
simple fix should be good enough.
Author: Forest Fang
Closes #9896 from saurfang/progressbar.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/800bd799
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/800bd799
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/800bd799
Branch: refs/heads/master
Commit: 800bd799acf7f10a469d8d6537279953129eb2c6
Parents: 12eea83
Author: Forest Fang
Authored: Tue Nov 24 09:03:32 2015 +
Committer: Sean Owen
Committed: Tue Nov 24 09:03:32 2015 +
--
core/src/main/scala/org/apache/spark/ui/UIUtils.scala | 4 +++-
.../src/test/scala/org/apache/spark/ui/UIUtilsSuite.scala | 10 ++
2 files changed, 13 insertions(+), 1 deletion(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/800bd799/core/src/main/scala/org/apache/spark/ui/UIUtils.scala
--
diff --git a/core/src/main/scala/org/apache/spark/ui/UIUtils.scala
b/core/src/main/scala/org/apache/spark/ui/UIUtils.scala
index 25dcb60..84a1116 100644
--- a/core/src/main/scala/org/apache/spark/ui/UIUtils.scala
+++ b/core/src/main/scala/org/apache/spark/ui/UIUtils.scala
@@ -319,7 +319,9 @@ private[spark] object UIUtils extends Logging {
skipped: Int,
total: Int): Seq[Node] = {
val completeWidth = "width: %s%%".format((completed.toDouble/total)*100)
-val startWidth = "width: %s%%".format((started.toDouble/total)*100)
+// started + completed can be > total when there are speculative tasks
+val boundedStarted = math.min(started, total - completed)
+val startWidth = "width: %s%%".format((boundedStarted.toDouble/total)*100)
http://git-wip-us.apache.org/repos/asf/spark/blob/800bd799/core/src/test/scala/org/apache/spark/ui/UIUtilsSuite.scala
--
diff --git a/core/src/test/scala/org/apache/spark/ui/UIUtilsSuite.scala
b/core/src/test/scala/org/apache/spark/ui/UIUtilsSuite.scala
index 2b693c1..dd8d5ec 100644
--- a/core/src/test/scala/org/apache/spark/ui/UIUtilsSuite.scala
+++ b/core/src/test/scala/org/apache/spark/ui/UIUtilsSuite.scala
@@ -57,6 +57,16 @@ class UIUtilsSuite extends SparkFunSuite {
)
}
+ test("SPARK-11906: Progress bar should not overflow because of speculative
tasks") {
+val generated = makeProgressBar(2, 3, 0, 0, 4).head.child.filter(_.label
== "div")
+val expected = Seq(
+ ,
+
+)
+assert(generated.sameElements(expected),
+ s"\nRunning progress bar should round
down\n\nExpected:\n$expected\nGenerated:\n$generated")
+ }
+
private def verify(
desc: String, expected: Elem, errorMsg: String = "", baseUrl: String =
""): Unit = {
val generated = makeDescription(desc, baseUrl)
-
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
spark git commit: [SPARK-11043][SQL] BugFix:Set the operator log in the thrift server.
Repository: spark
Updated Branches:
refs/heads/master 800bd799a -> d4a5e6f71
[SPARK-11043][SQL] BugFix:Set the operator log in the thrift server.
`SessionManager` will set the `operationLog` if the configuration
`hive.server2.logging.operation.enabled` is true in version of hive 1.2.1.
But the spark did not adapt to this change, so no matter enabled the
configuration or not, spark thrift server will always log the warn message.
PS: if `hive.server2.logging.operation.enabled` is false, it should log the
warn message (the same as hive thrift server).
Author: huangzhaowei
Closes #9056 from SaintBacchus/SPARK-11043.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/d4a5e6f7
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/d4a5e6f7
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/d4a5e6f7
Branch: refs/heads/master
Commit: d4a5e6f719079639ffd38470f4d8d1f6fde3228d
Parents: 800bd79
Author: huangzhaowei
Authored: Tue Nov 24 23:24:49 2015 +0800
Committer: Cheng Lian
Committed: Tue Nov 24 23:24:49 2015 +0800
--
.../SparkExecuteStatementOperation.scala| 8
.../hive/thriftserver/SparkSQLSessionManager.scala | 5 +
.../hive/thriftserver/HiveThriftServer2Suites.scala | 16 +++-
3 files changed, 24 insertions(+), 5 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/d4a5e6f7/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkExecuteStatementOperation.scala
--
diff --git
a/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkExecuteStatementOperation.scala
b/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkExecuteStatementOperation.scala
index 82fef92..e022ee8 100644
---
a/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkExecuteStatementOperation.scala
+++
b/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkExecuteStatementOperation.scala
@@ -134,12 +134,12 @@ private[hive] class SparkExecuteStatementOperation(
def getResultSetSchema: TableSchema = resultSchema
- override def run(): Unit = {
+ override def runInternal(): Unit = {
setState(OperationState.PENDING)
setHasResultSet(true) // avoid no resultset for async run
if (!runInBackground) {
- runInternal()
+ execute()
} else {
val sparkServiceUGI = Utils.getUGI()
@@ -151,7 +151,7 @@ private[hive] class SparkExecuteStatementOperation(
val doAsAction = new PrivilegedExceptionAction[Unit]() {
override def run(): Unit = {
try {
-runInternal()
+execute()
} catch {
case e: HiveSQLException =>
setOperationException(e)
@@ -188,7 +188,7 @@ private[hive] class SparkExecuteStatementOperation(
}
}
- override def runInternal(): Unit = {
+ private def execute(): Unit = {
statementId = UUID.randomUUID().toString
logInfo(s"Running query '$statement' with $statementId")
setState(OperationState.RUNNING)
http://git-wip-us.apache.org/repos/asf/spark/blob/d4a5e6f7/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkSQLSessionManager.scala
--
diff --git
a/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkSQLSessionManager.scala
b/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkSQLSessionManager.scala
index af4fcdf..de4e9c6 100644
---
a/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkSQLSessionManager.scala
+++
b/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkSQLSessionManager.scala
@@ -41,6 +41,11 @@ private[hive] class SparkSQLSessionManager(hiveServer:
HiveServer2, hiveContext:
override def init(hiveConf: HiveConf) {
setSuperField(this, "hiveConf", hiveConf)
+// Create operation log root directory, if operation logging is enabled
+if (hiveConf.getBoolVar(ConfVars.HIVE_SERVER2_LOGGING_OPERATION_ENABLED)) {
+ invoke(classOf[SessionManager], this, "initOperationLogRootDir")
+}
+
val backgroundPoolSize =
hiveConf.getIntVar(ConfVars.HIVE_SERVER2_ASYNC_EXEC_THREADS)
setSuperField(this, "backgroundOperationPool",
Executors.newFixedThreadPool(backgroundPoolSize))
getAncestorField[Log](this, 3, "LOG").info(
http://git-wip-us.apache.org/repos/asf/spark/blob/d4a5e6f7/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/HiveThriftServer2Suites.scala
spark git commit: [SPARK-11043][SQL] BugFix:Set the operator log in the thrift server.
Repository: spark
Updated Branches:
refs/heads/branch-1.6 17ea95133 -> f1f2cee4c
[SPARK-11043][SQL] BugFix:Set the operator log in the thrift server.
`SessionManager` will set the `operationLog` if the configuration
`hive.server2.logging.operation.enabled` is true in version of hive 1.2.1.
But the spark did not adapt to this change, so no matter enabled the
configuration or not, spark thrift server will always log the warn message.
PS: if `hive.server2.logging.operation.enabled` is false, it should log the
warn message (the same as hive thrift server).
Author: huangzhaowei
Closes #9056 from SaintBacchus/SPARK-11043.
(cherry picked from commit d4a5e6f719079639ffd38470f4d8d1f6fde3228d)
Signed-off-by: Cheng Lian
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/f1f2cee4
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/f1f2cee4
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/f1f2cee4
Branch: refs/heads/branch-1.6
Commit: f1f2cee4cebd624f86a3c59297aeb77e5a839e2a
Parents: 17ea951
Author: huangzhaowei
Authored: Tue Nov 24 23:24:49 2015 +0800
Committer: Cheng Lian
Committed: Tue Nov 24 23:25:14 2015 +0800
--
.../SparkExecuteStatementOperation.scala| 8
.../hive/thriftserver/SparkSQLSessionManager.scala | 5 +
.../hive/thriftserver/HiveThriftServer2Suites.scala | 16 +++-
3 files changed, 24 insertions(+), 5 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/f1f2cee4/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkExecuteStatementOperation.scala
--
diff --git
a/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkExecuteStatementOperation.scala
b/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkExecuteStatementOperation.scala
index 82fef92..e022ee8 100644
---
a/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkExecuteStatementOperation.scala
+++
b/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkExecuteStatementOperation.scala
@@ -134,12 +134,12 @@ private[hive] class SparkExecuteStatementOperation(
def getResultSetSchema: TableSchema = resultSchema
- override def run(): Unit = {
+ override def runInternal(): Unit = {
setState(OperationState.PENDING)
setHasResultSet(true) // avoid no resultset for async run
if (!runInBackground) {
- runInternal()
+ execute()
} else {
val sparkServiceUGI = Utils.getUGI()
@@ -151,7 +151,7 @@ private[hive] class SparkExecuteStatementOperation(
val doAsAction = new PrivilegedExceptionAction[Unit]() {
override def run(): Unit = {
try {
-runInternal()
+execute()
} catch {
case e: HiveSQLException =>
setOperationException(e)
@@ -188,7 +188,7 @@ private[hive] class SparkExecuteStatementOperation(
}
}
- override def runInternal(): Unit = {
+ private def execute(): Unit = {
statementId = UUID.randomUUID().toString
logInfo(s"Running query '$statement' with $statementId")
setState(OperationState.RUNNING)
http://git-wip-us.apache.org/repos/asf/spark/blob/f1f2cee4/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkSQLSessionManager.scala
--
diff --git
a/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkSQLSessionManager.scala
b/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkSQLSessionManager.scala
index af4fcdf..de4e9c6 100644
---
a/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkSQLSessionManager.scala
+++
b/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkSQLSessionManager.scala
@@ -41,6 +41,11 @@ private[hive] class SparkSQLSessionManager(hiveServer:
HiveServer2, hiveContext:
override def init(hiveConf: HiveConf) {
setSuperField(this, "hiveConf", hiveConf)
+// Create operation log root directory, if operation logging is enabled
+if (hiveConf.getBoolVar(ConfVars.HIVE_SERVER2_LOGGING_OPERATION_ENABLED)) {
+ invoke(classOf[SessionManager], this, "initOperationLogRootDir")
+}
+
val backgroundPoolSize =
hiveConf.getIntVar(ConfVars.HIVE_SERVER2_ASYNC_EXEC_THREADS)
setSuperField(this, "backgroundOperationPool",
Executors.newFixedThreadPool(backgroundPoolSize))
getAncestorField[Log](this, 3, "LOG").info(
http://git-wip-us.apache.org/repos/asf/spark/blob/f1f2cee4/sql/hive-thriftserver/src/test/scala/org/apache/spark
spark git commit: [SPARK-11592][SQL] flush spark-sql command line history to history file
Repository: spark
Updated Branches:
refs/heads/master d4a5e6f71 -> 5889880fb
[SPARK-11592][SQL] flush spark-sql command line history to history file
Currently, `spark-sql` would not flush command history when exiting.
Author: Daoyuan Wang
Closes #9563 from adrian-wang/jline.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/5889880f
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/5889880f
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/5889880f
Branch: refs/heads/master
Commit: 5889880fbe9628681042036892ef7ebd4f0857b4
Parents: d4a5e6f
Author: Daoyuan Wang
Authored: Tue Nov 24 23:32:05 2015 +0800
Committer: Cheng Lian
Committed: Tue Nov 24 23:32:05 2015 +0800
--
.../sql/hive/thriftserver/SparkSQLCLIDriver.scala | 16
1 file changed, 16 insertions(+)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/5889880f/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkSQLCLIDriver.scala
--
diff --git
a/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkSQLCLIDriver.scala
b/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkSQLCLIDriver.scala
index 6419002..4b928e6 100644
---
a/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkSQLCLIDriver.scala
+++
b/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkSQLCLIDriver.scala
@@ -194,6 +194,22 @@ private[hive] object SparkSQLCLIDriver extends Logging {
logWarning(e.getMessage)
}
+// add shutdown hook to flush the history to history file
+Runtime.getRuntime.addShutdownHook(new Thread(new Runnable() {
+ override def run() = {
+reader.getHistory match {
+ case h: FileHistory =>
+try {
+ h.flush()
+} catch {
+ case e: IOException =>
+logWarning("WARNING: Failed to write command history file: " +
e.getMessage)
+}
+ case _ =>
+}
+ }
+}))
+
// TODO: missing
/*
val clientTransportTSocketField =
classOf[CliSessionState].getDeclaredField("transport")
-
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
spark git commit: [SPARK-11818][REPL] Fix ExecutorClassLoader to lookup resources from …
Repository: spark
Updated Branches:
refs/heads/master 5889880fb -> be9dd1550
[SPARK-11818][REPL] Fix ExecutorClassLoader to lookup resources from â¦
â¦parent class loader
Without patch, two additional tests of ExecutorClassLoaderSuite fails.
- "resource from parent"
- "resources from parent"
Detailed explanation is here,
https://issues.apache.org/jira/browse/SPARK-11818?focusedCommentId=15011202&page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel#comment-15011202
Author: Jungtaek Lim
Closes #9812 from HeartSaVioR/SPARK-11818.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/be9dd155
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/be9dd155
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/be9dd155
Branch: refs/heads/master
Commit: be9dd1550c1816559d3d418a19c692e715f1c94e
Parents: 5889880
Author: Jungtaek Lim
Authored: Tue Nov 24 09:20:09 2015 -0800
Committer: Marcelo Vanzin
Committed: Tue Nov 24 09:20:09 2015 -0800
--
.../apache/spark/repl/ExecutorClassLoader.scala | 12 +++-
.../spark/repl/ExecutorClassLoaderSuite.scala | 29
2 files changed, 40 insertions(+), 1 deletion(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/be9dd155/repl/src/main/scala/org/apache/spark/repl/ExecutorClassLoader.scala
--
diff --git
a/repl/src/main/scala/org/apache/spark/repl/ExecutorClassLoader.scala
b/repl/src/main/scala/org/apache/spark/repl/ExecutorClassLoader.scala
index a976e96..a8859fc 100644
--- a/repl/src/main/scala/org/apache/spark/repl/ExecutorClassLoader.scala
+++ b/repl/src/main/scala/org/apache/spark/repl/ExecutorClassLoader.scala
@@ -34,7 +34,9 @@ import org.apache.spark.util.ParentClassLoader
/**
* A ClassLoader that reads classes from a Hadoop FileSystem or HTTP URI,
* used to load classes defined by the interpreter when the REPL is used.
- * Allows the user to specify if user class path should be first
+ * Allows the user to specify if user class path should be first.
+ * This class loader delegates getting/finding resources to parent loader,
+ * which makes sense until REPL never provide resource dynamically.
*/
class ExecutorClassLoader(conf: SparkConf, classUri: String, parent:
ClassLoader,
userClassPathFirst: Boolean) extends ClassLoader with Logging {
@@ -55,6 +57,14 @@ class ExecutorClassLoader(conf: SparkConf, classUri: String,
parent: ClassLoader
}
}
+ override def getResource(name: String): URL = {
+parentLoader.getResource(name)
+ }
+
+ override def getResources(name: String): java.util.Enumeration[URL] = {
+parentLoader.getResources(name)
+ }
+
override def findClass(name: String): Class[_] = {
userClassPathFirst match {
case true =>
findClassLocally(name).getOrElse(parentLoader.loadClass(name))
http://git-wip-us.apache.org/repos/asf/spark/blob/be9dd155/repl/src/test/scala/org/apache/spark/repl/ExecutorClassLoaderSuite.scala
--
diff --git
a/repl/src/test/scala/org/apache/spark/repl/ExecutorClassLoaderSuite.scala
b/repl/src/test/scala/org/apache/spark/repl/ExecutorClassLoaderSuite.scala
index a58eda1..c1211f7 100644
--- a/repl/src/test/scala/org/apache/spark/repl/ExecutorClassLoaderSuite.scala
+++ b/repl/src/test/scala/org/apache/spark/repl/ExecutorClassLoaderSuite.scala
@@ -19,8 +19,13 @@ package org.apache.spark.repl
import java.io.File
import java.net.{URL, URLClassLoader}
+import java.nio.charset.StandardCharsets
+import java.util
+
+import com.google.common.io.Files
import scala.concurrent.duration._
+import scala.io.Source
import scala.language.implicitConversions
import scala.language.postfixOps
@@ -41,6 +46,7 @@ class ExecutorClassLoaderSuite
val childClassNames = List("ReplFakeClass1", "ReplFakeClass2")
val parentClassNames = List("ReplFakeClass1", "ReplFakeClass2",
"ReplFakeClass3")
+ val parentResourceNames = List("fake-resource.txt")
var tempDir1: File = _
var tempDir2: File = _
var url1: String = _
@@ -54,6 +60,9 @@ class ExecutorClassLoaderSuite
url1 = "file://" + tempDir1
urls2 = List(tempDir2.toURI.toURL).toArray
childClassNames.foreach(TestUtils.createCompiledClass(_, tempDir1, "1"))
+parentResourceNames.foreach { x =>
+ Files.write("resource".getBytes(StandardCharsets.UTF_8), new
File(tempDir2, x))
+}
parentClassNames.foreach(TestUtils.createCompiledClass(_, tempDir2, "2"))
}
@@ -99,6 +108,26 @@ class ExecutorClassLoaderSuite
}
}
+ test("resource from parent") {
+val parentLoader = new URLClassLoader(urls2, null)
+val classLoader = new ExecutorClassLoader(new SparkConf(), url1,
parentLoader, true)
+val resourceName
spark git commit: [SPARK-11818][REPL] Fix ExecutorClassLoader to lookup resources from …
Repository: spark
Updated Branches:
refs/heads/branch-1.6 f1f2cee4c -> 895128505
[SPARK-11818][REPL] Fix ExecutorClassLoader to lookup resources from â¦
â¦parent class loader
Without patch, two additional tests of ExecutorClassLoaderSuite fails.
- "resource from parent"
- "resources from parent"
Detailed explanation is here,
https://issues.apache.org/jira/browse/SPARK-11818?focusedCommentId=15011202&page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel#comment-15011202
Author: Jungtaek Lim
Closes #9812 from HeartSaVioR/SPARK-11818.
(cherry picked from commit be9dd1550c1816559d3d418a19c692e715f1c94e)
Signed-off-by: Marcelo Vanzin
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/89512850
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/89512850
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/89512850
Branch: refs/heads/branch-1.6
Commit: 8951285054c4a4496b1c736fe225491d6d8d3426
Parents: f1f2cee
Author: Jungtaek Lim
Authored: Tue Nov 24 09:20:09 2015 -0800
Committer: Marcelo Vanzin
Committed: Tue Nov 24 09:20:24 2015 -0800
--
.../apache/spark/repl/ExecutorClassLoader.scala | 12 +++-
.../spark/repl/ExecutorClassLoaderSuite.scala | 29
2 files changed, 40 insertions(+), 1 deletion(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/89512850/repl/src/main/scala/org/apache/spark/repl/ExecutorClassLoader.scala
--
diff --git
a/repl/src/main/scala/org/apache/spark/repl/ExecutorClassLoader.scala
b/repl/src/main/scala/org/apache/spark/repl/ExecutorClassLoader.scala
index a976e96..a8859fc 100644
--- a/repl/src/main/scala/org/apache/spark/repl/ExecutorClassLoader.scala
+++ b/repl/src/main/scala/org/apache/spark/repl/ExecutorClassLoader.scala
@@ -34,7 +34,9 @@ import org.apache.spark.util.ParentClassLoader
/**
* A ClassLoader that reads classes from a Hadoop FileSystem or HTTP URI,
* used to load classes defined by the interpreter when the REPL is used.
- * Allows the user to specify if user class path should be first
+ * Allows the user to specify if user class path should be first.
+ * This class loader delegates getting/finding resources to parent loader,
+ * which makes sense until REPL never provide resource dynamically.
*/
class ExecutorClassLoader(conf: SparkConf, classUri: String, parent:
ClassLoader,
userClassPathFirst: Boolean) extends ClassLoader with Logging {
@@ -55,6 +57,14 @@ class ExecutorClassLoader(conf: SparkConf, classUri: String,
parent: ClassLoader
}
}
+ override def getResource(name: String): URL = {
+parentLoader.getResource(name)
+ }
+
+ override def getResources(name: String): java.util.Enumeration[URL] = {
+parentLoader.getResources(name)
+ }
+
override def findClass(name: String): Class[_] = {
userClassPathFirst match {
case true =>
findClassLocally(name).getOrElse(parentLoader.loadClass(name))
http://git-wip-us.apache.org/repos/asf/spark/blob/89512850/repl/src/test/scala/org/apache/spark/repl/ExecutorClassLoaderSuite.scala
--
diff --git
a/repl/src/test/scala/org/apache/spark/repl/ExecutorClassLoaderSuite.scala
b/repl/src/test/scala/org/apache/spark/repl/ExecutorClassLoaderSuite.scala
index a58eda1..c1211f7 100644
--- a/repl/src/test/scala/org/apache/spark/repl/ExecutorClassLoaderSuite.scala
+++ b/repl/src/test/scala/org/apache/spark/repl/ExecutorClassLoaderSuite.scala
@@ -19,8 +19,13 @@ package org.apache.spark.repl
import java.io.File
import java.net.{URL, URLClassLoader}
+import java.nio.charset.StandardCharsets
+import java.util
+
+import com.google.common.io.Files
import scala.concurrent.duration._
+import scala.io.Source
import scala.language.implicitConversions
import scala.language.postfixOps
@@ -41,6 +46,7 @@ class ExecutorClassLoaderSuite
val childClassNames = List("ReplFakeClass1", "ReplFakeClass2")
val parentClassNames = List("ReplFakeClass1", "ReplFakeClass2",
"ReplFakeClass3")
+ val parentResourceNames = List("fake-resource.txt")
var tempDir1: File = _
var tempDir2: File = _
var url1: String = _
@@ -54,6 +60,9 @@ class ExecutorClassLoaderSuite
url1 = "file://" + tempDir1
urls2 = List(tempDir2.toURI.toURL).toArray
childClassNames.foreach(TestUtils.createCompiledClass(_, tempDir1, "1"))
+parentResourceNames.foreach { x =>
+ Files.write("resource".getBytes(StandardCharsets.UTF_8), new
File(tempDir2, x))
+}
parentClassNames.foreach(TestUtils.createCompiledClass(_, tempDir2, "2"))
}
@@ -99,6 +108,26 @@ class ExecutorClassLoaderSuite
}
}
+ test("resource from parent") {
+val parentLoader = new URLClassLoader(urls2, null)
+
spark git commit: [SPARK-11942][SQL] fix encoder life cycle for CoGroup
Repository: spark
Updated Branches:
refs/heads/branch-1.6 895128505 -> 3cb1b6d39
[SPARK-11942][SQL] fix encoder life cycle for CoGroup
we should pass in resolved encodera to logical `CoGroup` and bind them in
physical `CoGroup`
Author: Wenchen Fan
Closes #9928 from cloud-fan/cogroup.
(cherry picked from commit e5aaae6e1145b8c25c4872b2992ab425da9c6f9b)
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/3cb1b6d3
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/3cb1b6d3
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/3cb1b6d3
Branch: refs/heads/branch-1.6
Commit: 3cb1b6d398a8066a6c9d800a16b8cd21f3fde50f
Parents: 8951285
Author: Wenchen Fan
Authored: Tue Nov 24 09:28:39 2015 -0800
Committer: Michael Armbrust
Committed: Tue Nov 24 09:28:51 2015 -0800
--
.../catalyst/plans/logical/basicOperators.scala | 27 +++-
.../org/apache/spark/sql/GroupedDataset.scala | 4 ++-
.../spark/sql/execution/basicOperators.scala| 20 ---
.../org/apache/spark/sql/DatasetSuite.scala | 12 +
4 files changed, 41 insertions(+), 22 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/3cb1b6d3/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicOperators.scala
--
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicOperators.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicOperators.scala
index 737e62f..5665fd7 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicOperators.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicOperators.scala
@@ -553,19 +553,22 @@ case class MapGroups[K, T, U](
/** Factory for constructing new `CoGroup` nodes. */
object CoGroup {
- def apply[K : Encoder, Left : Encoder, Right : Encoder, R : Encoder](
- func: (K, Iterator[Left], Iterator[Right]) => TraversableOnce[R],
+ def apply[Key, Left, Right, Result : Encoder](
+ func: (Key, Iterator[Left], Iterator[Right]) => TraversableOnce[Result],
+ keyEnc: ExpressionEncoder[Key],
+ leftEnc: ExpressionEncoder[Left],
+ rightEnc: ExpressionEncoder[Right],
leftGroup: Seq[Attribute],
rightGroup: Seq[Attribute],
left: LogicalPlan,
- right: LogicalPlan): CoGroup[K, Left, Right, R] = {
+ right: LogicalPlan): CoGroup[Key, Left, Right, Result] = {
CoGroup(
func,
- encoderFor[K],
- encoderFor[Left],
- encoderFor[Right],
- encoderFor[R],
- encoderFor[R].schema.toAttributes,
+ keyEnc,
+ leftEnc,
+ rightEnc,
+ encoderFor[Result],
+ encoderFor[Result].schema.toAttributes,
leftGroup,
rightGroup,
left,
@@ -577,12 +580,12 @@ object CoGroup {
* A relation produced by applying `func` to each grouping key and associated
values from left and
* right children.
*/
-case class CoGroup[K, Left, Right, R](
-func: (K, Iterator[Left], Iterator[Right]) => TraversableOnce[R],
-kEncoder: ExpressionEncoder[K],
+case class CoGroup[Key, Left, Right, Result](
+func: (Key, Iterator[Left], Iterator[Right]) => TraversableOnce[Result],
+keyEnc: ExpressionEncoder[Key],
leftEnc: ExpressionEncoder[Left],
rightEnc: ExpressionEncoder[Right],
-rEncoder: ExpressionEncoder[R],
+resultEnc: ExpressionEncoder[Result],
output: Seq[Attribute],
leftGroup: Seq[Attribute],
rightGroup: Seq[Attribute],
http://git-wip-us.apache.org/repos/asf/spark/blob/3cb1b6d3/sql/core/src/main/scala/org/apache/spark/sql/GroupedDataset.scala
--
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/GroupedDataset.scala
b/sql/core/src/main/scala/org/apache/spark/sql/GroupedDataset.scala
index 793a86b..a10a893 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/GroupedDataset.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/GroupedDataset.scala
@@ -304,11 +304,13 @@ class GroupedDataset[K, V] private[sql](
def cogroup[U, R : Encoder](
other: GroupedDataset[K, U])(
f: (K, Iterator[V], Iterator[U]) => TraversableOnce[R]): Dataset[R] = {
-implicit def uEnc: Encoder[U] = other.unresolvedVEncoder
new Dataset[R](
sqlContext,
CoGroup(
f,
+this.resolvedKEncoder,
+this.resolvedVEncoder,
+other.resolvedVEncoder,
this.groupingAttributes,
other.groupingAttributes,
this.logicalPlan,
http://git-wip-us.apache.org/repos/asf/spark/blob/3cb1b6d3/sql/core/src/main/scala/org/apache/spark/sql/execution/basicOperato
spark git commit: [SPARK-11942][SQL] fix encoder life cycle for CoGroup
Repository: spark
Updated Branches:
refs/heads/master be9dd1550 -> e5aaae6e1
[SPARK-11942][SQL] fix encoder life cycle for CoGroup
we should pass in resolved encodera to logical `CoGroup` and bind them in
physical `CoGroup`
Author: Wenchen Fan
Closes #9928 from cloud-fan/cogroup.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/e5aaae6e
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/e5aaae6e
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/e5aaae6e
Branch: refs/heads/master
Commit: e5aaae6e1145b8c25c4872b2992ab425da9c6f9b
Parents: be9dd15
Author: Wenchen Fan
Authored: Tue Nov 24 09:28:39 2015 -0800
Committer: Michael Armbrust
Committed: Tue Nov 24 09:28:39 2015 -0800
--
.../catalyst/plans/logical/basicOperators.scala | 27 +++-
.../org/apache/spark/sql/GroupedDataset.scala | 4 ++-
.../spark/sql/execution/basicOperators.scala| 20 ---
.../org/apache/spark/sql/DatasetSuite.scala | 12 +
4 files changed, 41 insertions(+), 22 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/e5aaae6e/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicOperators.scala
--
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicOperators.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicOperators.scala
index 737e62f..5665fd7 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicOperators.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicOperators.scala
@@ -553,19 +553,22 @@ case class MapGroups[K, T, U](
/** Factory for constructing new `CoGroup` nodes. */
object CoGroup {
- def apply[K : Encoder, Left : Encoder, Right : Encoder, R : Encoder](
- func: (K, Iterator[Left], Iterator[Right]) => TraversableOnce[R],
+ def apply[Key, Left, Right, Result : Encoder](
+ func: (Key, Iterator[Left], Iterator[Right]) => TraversableOnce[Result],
+ keyEnc: ExpressionEncoder[Key],
+ leftEnc: ExpressionEncoder[Left],
+ rightEnc: ExpressionEncoder[Right],
leftGroup: Seq[Attribute],
rightGroup: Seq[Attribute],
left: LogicalPlan,
- right: LogicalPlan): CoGroup[K, Left, Right, R] = {
+ right: LogicalPlan): CoGroup[Key, Left, Right, Result] = {
CoGroup(
func,
- encoderFor[K],
- encoderFor[Left],
- encoderFor[Right],
- encoderFor[R],
- encoderFor[R].schema.toAttributes,
+ keyEnc,
+ leftEnc,
+ rightEnc,
+ encoderFor[Result],
+ encoderFor[Result].schema.toAttributes,
leftGroup,
rightGroup,
left,
@@ -577,12 +580,12 @@ object CoGroup {
* A relation produced by applying `func` to each grouping key and associated
values from left and
* right children.
*/
-case class CoGroup[K, Left, Right, R](
-func: (K, Iterator[Left], Iterator[Right]) => TraversableOnce[R],
-kEncoder: ExpressionEncoder[K],
+case class CoGroup[Key, Left, Right, Result](
+func: (Key, Iterator[Left], Iterator[Right]) => TraversableOnce[Result],
+keyEnc: ExpressionEncoder[Key],
leftEnc: ExpressionEncoder[Left],
rightEnc: ExpressionEncoder[Right],
-rEncoder: ExpressionEncoder[R],
+resultEnc: ExpressionEncoder[Result],
output: Seq[Attribute],
leftGroup: Seq[Attribute],
rightGroup: Seq[Attribute],
http://git-wip-us.apache.org/repos/asf/spark/blob/e5aaae6e/sql/core/src/main/scala/org/apache/spark/sql/GroupedDataset.scala
--
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/GroupedDataset.scala
b/sql/core/src/main/scala/org/apache/spark/sql/GroupedDataset.scala
index 793a86b..a10a893 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/GroupedDataset.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/GroupedDataset.scala
@@ -304,11 +304,13 @@ class GroupedDataset[K, V] private[sql](
def cogroup[U, R : Encoder](
other: GroupedDataset[K, U])(
f: (K, Iterator[V], Iterator[U]) => TraversableOnce[R]): Dataset[R] = {
-implicit def uEnc: Encoder[U] = other.unresolvedVEncoder
new Dataset[R](
sqlContext,
CoGroup(
f,
+this.resolvedKEncoder,
+this.resolvedVEncoder,
+other.resolvedVEncoder,
this.groupingAttributes,
other.groupingAttributes,
this.logicalPlan,
http://git-wip-us.apache.org/repos/asf/spark/blob/e5aaae6e/sql/core/src/main/scala/org/apache/spark/sql/execution/basicOperators.scala
--
diff --git
a/sql/core/src/main
spark git commit: [SPARK-11952][ML] Remove duplicate ml examples
Repository: spark
Updated Branches:
refs/heads/master e5aaae6e1 -> 56a0aba0a
[SPARK-11952][ML] Remove duplicate ml examples
Remove duplicate ml examples (only for ml). mengxr
Author: Yanbo Liang
Closes #9933 from yanboliang/SPARK-11685.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/56a0aba0
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/56a0aba0
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/56a0aba0
Branch: refs/heads/master
Commit: 56a0aba0a60326ba026056c9a23f3f6ec7258c19
Parents: e5aaae6
Author: Yanbo Liang
Authored: Tue Nov 24 09:52:53 2015 -0800
Committer: Xiangrui Meng
Committed: Tue Nov 24 09:52:53 2015 -0800
--
.../main/python/ml/gradient_boosted_trees.py| 82 --
.../src/main/python/ml/logistic_regression.py | 66 ---
.../src/main/python/ml/random_forest_example.py | 87
3 files changed, 235 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/56a0aba0/examples/src/main/python/ml/gradient_boosted_trees.py
--
diff --git a/examples/src/main/python/ml/gradient_boosted_trees.py
b/examples/src/main/python/ml/gradient_boosted_trees.py
deleted file mode 100644
index c3bf8aa..000
--- a/examples/src/main/python/ml/gradient_boosted_trees.py
+++ /dev/null
@@ -1,82 +0,0 @@
-#
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements. See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You under the Apache License, Version 2.0
-# (the "License"); you may not use this file except in compliance with
-# the License. You may obtain a copy of the License at
-#
-#http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-from __future__ import print_function
-
-import sys
-
-from pyspark import SparkContext
-from pyspark.ml.classification import GBTClassifier
-from pyspark.ml.feature import StringIndexer
-from pyspark.ml.regression import GBTRegressor
-from pyspark.mllib.evaluation import BinaryClassificationMetrics,
RegressionMetrics
-from pyspark.sql import Row, SQLContext
-
-"""
-A simple example demonstrating a Gradient Boosted Trees
Classification/Regression Pipeline.
-Note: GBTClassifier only supports binary classification currently
-Run with:
- bin/spark-submit examples/src/main/python/ml/gradient_boosted_trees.py
-"""
-
-
-def testClassification(train, test):
-# Train a GradientBoostedTrees model.
-
-rf = GBTClassifier(maxIter=30, maxDepth=4, labelCol="indexedLabel")
-
-model = rf.fit(train)
-predictionAndLabels = model.transform(test).select("prediction",
"indexedLabel") \
-.map(lambda x: (x.prediction, x.indexedLabel))
-
-metrics = BinaryClassificationMetrics(predictionAndLabels)
-print("AUC %.3f" % metrics.areaUnderROC)
-
-
-def testRegression(train, test):
-# Train a GradientBoostedTrees model.
-
-rf = GBTRegressor(maxIter=30, maxDepth=4, labelCol="indexedLabel")
-
-model = rf.fit(train)
-predictionAndLabels = model.transform(test).select("prediction",
"indexedLabel") \
-.map(lambda x: (x.prediction, x.indexedLabel))
-
-metrics = RegressionMetrics(predictionAndLabels)
-print("rmse %.3f" % metrics.rootMeanSquaredError)
-print("r2 %.3f" % metrics.r2)
-print("mae %.3f" % metrics.meanAbsoluteError)
-
-
-if __name__ == "__main__":
-if len(sys.argv) > 1:
-print("Usage: gradient_boosted_trees", file=sys.stderr)
-exit(1)
-sc = SparkContext(appName="PythonGBTExample")
-sqlContext = SQLContext(sc)
-
-# Load the data stored in LIBSVM format as a DataFrame.
-df =
sqlContext.read.format("libsvm").load("data/mllib/sample_libsvm_data.txt")
-
-# Map labels into an indexed column of labels in [0, numLabels)
-stringIndexer = StringIndexer(inputCol="label", outputCol="indexedLabel")
-si_model = stringIndexer.fit(df)
-td = si_model.transform(df)
-[train, test] = td.randomSplit([0.7, 0.3])
-testClassification(train, test)
-testRegression(train, test)
-sc.stop()
http://git-wip-us.apache.org/repos/asf/spark/blob/56a0aba0/examples/src/main/python/ml/logistic_regression.py
--
diff --git a/examples/src/main/python/ml/logistic_regression.py
b/examples/src/main/python/ml/logistic_regression.py
de
spark git commit: [SPARK-11952][ML] Remove duplicate ml examples
Repository: spark
Updated Branches:
refs/heads/branch-1.6 3cb1b6d39 -> 6914b7504
[SPARK-11952][ML] Remove duplicate ml examples
Remove duplicate ml examples (only for ml). mengxr
Author: Yanbo Liang
Closes #9933 from yanboliang/SPARK-11685.
(cherry picked from commit 56a0aba0a60326ba026056c9a23f3f6ec7258c19)
Signed-off-by: Xiangrui Meng
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/6914b750
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/6914b750
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/6914b750
Branch: refs/heads/branch-1.6
Commit: 6914b75046dceda47ba3ff904e67f55752e8d49d
Parents: 3cb1b6d
Author: Yanbo Liang
Authored: Tue Nov 24 09:52:53 2015 -0800
Committer: Xiangrui Meng
Committed: Tue Nov 24 09:53:01 2015 -0800
--
.../main/python/ml/gradient_boosted_trees.py| 82 --
.../src/main/python/ml/logistic_regression.py | 66 ---
.../src/main/python/ml/random_forest_example.py | 87
3 files changed, 235 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/6914b750/examples/src/main/python/ml/gradient_boosted_trees.py
--
diff --git a/examples/src/main/python/ml/gradient_boosted_trees.py
b/examples/src/main/python/ml/gradient_boosted_trees.py
deleted file mode 100644
index c3bf8aa..000
--- a/examples/src/main/python/ml/gradient_boosted_trees.py
+++ /dev/null
@@ -1,82 +0,0 @@
-#
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements. See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You under the Apache License, Version 2.0
-# (the "License"); you may not use this file except in compliance with
-# the License. You may obtain a copy of the License at
-#
-#http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-from __future__ import print_function
-
-import sys
-
-from pyspark import SparkContext
-from pyspark.ml.classification import GBTClassifier
-from pyspark.ml.feature import StringIndexer
-from pyspark.ml.regression import GBTRegressor
-from pyspark.mllib.evaluation import BinaryClassificationMetrics,
RegressionMetrics
-from pyspark.sql import Row, SQLContext
-
-"""
-A simple example demonstrating a Gradient Boosted Trees
Classification/Regression Pipeline.
-Note: GBTClassifier only supports binary classification currently
-Run with:
- bin/spark-submit examples/src/main/python/ml/gradient_boosted_trees.py
-"""
-
-
-def testClassification(train, test):
-# Train a GradientBoostedTrees model.
-
-rf = GBTClassifier(maxIter=30, maxDepth=4, labelCol="indexedLabel")
-
-model = rf.fit(train)
-predictionAndLabels = model.transform(test).select("prediction",
"indexedLabel") \
-.map(lambda x: (x.prediction, x.indexedLabel))
-
-metrics = BinaryClassificationMetrics(predictionAndLabels)
-print("AUC %.3f" % metrics.areaUnderROC)
-
-
-def testRegression(train, test):
-# Train a GradientBoostedTrees model.
-
-rf = GBTRegressor(maxIter=30, maxDepth=4, labelCol="indexedLabel")
-
-model = rf.fit(train)
-predictionAndLabels = model.transform(test).select("prediction",
"indexedLabel") \
-.map(lambda x: (x.prediction, x.indexedLabel))
-
-metrics = RegressionMetrics(predictionAndLabels)
-print("rmse %.3f" % metrics.rootMeanSquaredError)
-print("r2 %.3f" % metrics.r2)
-print("mae %.3f" % metrics.meanAbsoluteError)
-
-
-if __name__ == "__main__":
-if len(sys.argv) > 1:
-print("Usage: gradient_boosted_trees", file=sys.stderr)
-exit(1)
-sc = SparkContext(appName="PythonGBTExample")
-sqlContext = SQLContext(sc)
-
-# Load the data stored in LIBSVM format as a DataFrame.
-df =
sqlContext.read.format("libsvm").load("data/mllib/sample_libsvm_data.txt")
-
-# Map labels into an indexed column of labels in [0, numLabels)
-stringIndexer = StringIndexer(inputCol="label", outputCol="indexedLabel")
-si_model = stringIndexer.fit(df)
-td = si_model.transform(df)
-[train, test] = td.randomSplit([0.7, 0.3])
-testClassification(train, test)
-testRegression(train, test)
-sc.stop()
http://git-wip-us.apache.org/repos/asf/spark/blob/6914b750/examples/src/main/python/ml/logistic_regression.py
--
diff --git a
spark git commit: [SPARK-11521][ML][DOC] Document that Logistic, Linear Regression summaries ignore weight col
Repository: spark
Updated Branches:
refs/heads/master 56a0aba0a -> 9e24ba667
[SPARK-11521][ML][DOC] Document that Logistic, Linear Regression summaries
ignore weight col
Doc for 1.6 that the summaries mostly ignore the weight column.
To be corrected for 1.7
CC: mengxr thunterdb
Author: Joseph K. Bradley
Closes #9927 from jkbradley/linregsummary-doc.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/9e24ba66
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/9e24ba66
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/9e24ba66
Branch: refs/heads/master
Commit: 9e24ba667e43290fbaa3cacb93cf5d9be790f1fd
Parents: 56a0aba
Author: Joseph K. Bradley
Authored: Tue Nov 24 09:54:55 2015 -0800
Committer: Xiangrui Meng
Committed: Tue Nov 24 09:54:55 2015 -0800
--
.../ml/classification/LogisticRegression.scala| 18 ++
.../spark/ml/regression/LinearRegression.scala| 15 +++
2 files changed, 33 insertions(+)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/9e24ba66/mllib/src/main/scala/org/apache/spark/ml/classification/LogisticRegression.scala
--
diff --git
a/mllib/src/main/scala/org/apache/spark/ml/classification/LogisticRegression.scala
b/mllib/src/main/scala/org/apache/spark/ml/classification/LogisticRegression.scala
index 418bbdc..d320d64 100644
---
a/mllib/src/main/scala/org/apache/spark/ml/classification/LogisticRegression.scala
+++
b/mllib/src/main/scala/org/apache/spark/ml/classification/LogisticRegression.scala
@@ -755,23 +755,35 @@ class BinaryLogisticRegressionSummary
private[classification] (
* Returns the receiver operating characteristic (ROC) curve,
* which is an Dataframe having two fields (FPR, TPR)
* with (0.0, 0.0) prepended and (1.0, 1.0) appended to it.
+ *
+ * Note: This ignores instance weights (setting all to 1.0) from
[[LogisticRegression.weightCol]].
+ * This will change in later Spark versions.
* @see http://en.wikipedia.org/wiki/Receiver_operating_characteristic
*/
@transient lazy val roc: DataFrame = binaryMetrics.roc().toDF("FPR", "TPR")
/**
* Computes the area under the receiver operating characteristic (ROC) curve.
+ *
+ * Note: This ignores instance weights (setting all to 1.0) from
[[LogisticRegression.weightCol]].
+ * This will change in later Spark versions.
*/
lazy val areaUnderROC: Double = binaryMetrics.areaUnderROC()
/**
* Returns the precision-recall curve, which is an Dataframe containing
* two fields recall, precision with (0.0, 1.0) prepended to it.
+ *
+ * Note: This ignores instance weights (setting all to 1.0) from
[[LogisticRegression.weightCol]].
+ * This will change in later Spark versions.
*/
@transient lazy val pr: DataFrame = binaryMetrics.pr().toDF("recall",
"precision")
/**
* Returns a dataframe with two fields (threshold, F-Measure) curve with
beta = 1.0.
+ *
+ * Note: This ignores instance weights (setting all to 1.0) from
[[LogisticRegression.weightCol]].
+ * This will change in later Spark versions.
*/
@transient lazy val fMeasureByThreshold: DataFrame = {
binaryMetrics.fMeasureByThreshold().toDF("threshold", "F-Measure")
@@ -781,6 +793,9 @@ class BinaryLogisticRegressionSummary
private[classification] (
* Returns a dataframe with two fields (threshold, precision) curve.
* Every possible probability obtained in transforming the dataset are used
* as thresholds used in calculating the precision.
+ *
+ * Note: This ignores instance weights (setting all to 1.0) from
[[LogisticRegression.weightCol]].
+ * This will change in later Spark versions.
*/
@transient lazy val precisionByThreshold: DataFrame = {
binaryMetrics.precisionByThreshold().toDF("threshold", "precision")
@@ -790,6 +805,9 @@ class BinaryLogisticRegressionSummary
private[classification] (
* Returns a dataframe with two fields (threshold, recall) curve.
* Every possible probability obtained in transforming the dataset are used
* as thresholds used in calculating the recall.
+ *
+ * Note: This ignores instance weights (setting all to 1.0) from
[[LogisticRegression.weightCol]].
+ * This will change in later Spark versions.
*/
@transient lazy val recallByThreshold: DataFrame = {
binaryMetrics.recallByThreshold().toDF("threshold", "recall")
http://git-wip-us.apache.org/repos/asf/spark/blob/9e24ba66/mllib/src/main/scala/org/apache/spark/ml/regression/LinearRegression.scala
--
diff --git
a/mllib/src/main/scala/org/apache/spark/ml/regression/LinearRegression.scala
b/mllib/src/main/scala/org/apac
spark git commit: [SPARK-11521][ML][DOC] Document that Logistic, Linear Regression summaries ignore weight col
Repository: spark
Updated Branches:
refs/heads/branch-1.6 6914b7504 -> 70febe224
[SPARK-11521][ML][DOC] Document that Logistic, Linear Regression summaries
ignore weight col
Doc for 1.6 that the summaries mostly ignore the weight column.
To be corrected for 1.7
CC: mengxr thunterdb
Author: Joseph K. Bradley
Closes #9927 from jkbradley/linregsummary-doc.
(cherry picked from commit 9e24ba667e43290fbaa3cacb93cf5d9be790f1fd)
Signed-off-by: Xiangrui Meng
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/70febe22
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/70febe22
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/70febe22
Branch: refs/heads/branch-1.6
Commit: 70febe224f64cb6468c14d4788a63b35d0475d41
Parents: 6914b75
Author: Joseph K. Bradley
Authored: Tue Nov 24 09:54:55 2015 -0800
Committer: Xiangrui Meng
Committed: Tue Nov 24 09:55:02 2015 -0800
--
.../ml/classification/LogisticRegression.scala| 18 ++
.../spark/ml/regression/LinearRegression.scala| 15 +++
2 files changed, 33 insertions(+)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/70febe22/mllib/src/main/scala/org/apache/spark/ml/classification/LogisticRegression.scala
--
diff --git
a/mllib/src/main/scala/org/apache/spark/ml/classification/LogisticRegression.scala
b/mllib/src/main/scala/org/apache/spark/ml/classification/LogisticRegression.scala
index 418bbdc..d320d64 100644
---
a/mllib/src/main/scala/org/apache/spark/ml/classification/LogisticRegression.scala
+++
b/mllib/src/main/scala/org/apache/spark/ml/classification/LogisticRegression.scala
@@ -755,23 +755,35 @@ class BinaryLogisticRegressionSummary
private[classification] (
* Returns the receiver operating characteristic (ROC) curve,
* which is an Dataframe having two fields (FPR, TPR)
* with (0.0, 0.0) prepended and (1.0, 1.0) appended to it.
+ *
+ * Note: This ignores instance weights (setting all to 1.0) from
[[LogisticRegression.weightCol]].
+ * This will change in later Spark versions.
* @see http://en.wikipedia.org/wiki/Receiver_operating_characteristic
*/
@transient lazy val roc: DataFrame = binaryMetrics.roc().toDF("FPR", "TPR")
/**
* Computes the area under the receiver operating characteristic (ROC) curve.
+ *
+ * Note: This ignores instance weights (setting all to 1.0) from
[[LogisticRegression.weightCol]].
+ * This will change in later Spark versions.
*/
lazy val areaUnderROC: Double = binaryMetrics.areaUnderROC()
/**
* Returns the precision-recall curve, which is an Dataframe containing
* two fields recall, precision with (0.0, 1.0) prepended to it.
+ *
+ * Note: This ignores instance weights (setting all to 1.0) from
[[LogisticRegression.weightCol]].
+ * This will change in later Spark versions.
*/
@transient lazy val pr: DataFrame = binaryMetrics.pr().toDF("recall",
"precision")
/**
* Returns a dataframe with two fields (threshold, F-Measure) curve with
beta = 1.0.
+ *
+ * Note: This ignores instance weights (setting all to 1.0) from
[[LogisticRegression.weightCol]].
+ * This will change in later Spark versions.
*/
@transient lazy val fMeasureByThreshold: DataFrame = {
binaryMetrics.fMeasureByThreshold().toDF("threshold", "F-Measure")
@@ -781,6 +793,9 @@ class BinaryLogisticRegressionSummary
private[classification] (
* Returns a dataframe with two fields (threshold, precision) curve.
* Every possible probability obtained in transforming the dataset are used
* as thresholds used in calculating the precision.
+ *
+ * Note: This ignores instance weights (setting all to 1.0) from
[[LogisticRegression.weightCol]].
+ * This will change in later Spark versions.
*/
@transient lazy val precisionByThreshold: DataFrame = {
binaryMetrics.precisionByThreshold().toDF("threshold", "precision")
@@ -790,6 +805,9 @@ class BinaryLogisticRegressionSummary
private[classification] (
* Returns a dataframe with two fields (threshold, recall) curve.
* Every possible probability obtained in transforming the dataset are used
* as thresholds used in calculating the recall.
+ *
+ * Note: This ignores instance weights (setting all to 1.0) from
[[LogisticRegression.weightCol]].
+ * This will change in later Spark versions.
*/
@transient lazy val recallByThreshold: DataFrame = {
binaryMetrics.recallByThreshold().toDF("threshold", "recall")
http://git-wip-us.apache.org/repos/asf/spark/blob/70febe22/mllib/src/main/scala/org/apache/spark/ml/regression/LinearRegression.scala
--
diff --git
a
spark git commit: [SPARK-11847][ML] Model export/import for spark.ml: LDA
Repository: spark
Updated Branches:
refs/heads/branch-1.6 70febe224 -> af86c38db
[SPARK-11847][ML] Model export/import for spark.ml: LDA
Add read/write support to LDA, similar to ALS.
save/load for ml.LocalLDAModel is done.
For DistributedLDAModel, I'm not sure if we can invoke save on the
mllib.DistributedLDAModel directly. I'll send update after some test.
Author: Yuhao Yang
Closes #9894 from hhbyyh/ldaMLsave.
(cherry picked from commit 52bc25c8e26d4be250d8ff7864067528f4f98592)
Signed-off-by: Xiangrui Meng
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/af86c38d
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/af86c38d
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/af86c38d
Branch: refs/heads/branch-1.6
Commit: af86c38db7676c4dfc2724d5f86f0f5f3a22e349
Parents: 70febe2
Author: Yuhao Yang
Authored: Tue Nov 24 09:56:17 2015 -0800
Committer: Xiangrui Meng
Committed: Tue Nov 24 09:56:24 2015 -0800
--
.../org/apache/spark/ml/clustering/LDA.scala| 110 ++-
.../spark/mllib/clustering/LDAModel.scala | 4 +-
.../apache/spark/ml/clustering/LDASuite.scala | 44 +++-
3 files changed, 150 insertions(+), 8 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/af86c38d/mllib/src/main/scala/org/apache/spark/ml/clustering/LDA.scala
--
diff --git a/mllib/src/main/scala/org/apache/spark/ml/clustering/LDA.scala
b/mllib/src/main/scala/org/apache/spark/ml/clustering/LDA.scala
index 92e0581..830510b 100644
--- a/mllib/src/main/scala/org/apache/spark/ml/clustering/LDA.scala
+++ b/mllib/src/main/scala/org/apache/spark/ml/clustering/LDA.scala
@@ -17,12 +17,13 @@
package org.apache.spark.ml.clustering
+import org.apache.hadoop.fs.Path
import org.apache.spark.Logging
import org.apache.spark.annotation.{Experimental, Since}
-import org.apache.spark.ml.util.{SchemaUtils, Identifiable}
import org.apache.spark.ml.{Estimator, Model}
import org.apache.spark.ml.param.shared.{HasCheckpointInterval,
HasFeaturesCol, HasSeed, HasMaxIter}
import org.apache.spark.ml.param._
+import org.apache.spark.ml.util._
import org.apache.spark.mllib.clustering.{DistributedLDAModel =>
OldDistributedLDAModel,
EMLDAOptimizer => OldEMLDAOptimizer, LDA => OldLDA, LDAModel =>
OldLDAModel,
LDAOptimizer => OldLDAOptimizer, LocalLDAModel => OldLocalLDAModel,
@@ -322,7 +323,7 @@ sealed abstract class LDAModel private[ml] (
@Since("1.6.0") override val uid: String,
@Since("1.6.0") val vocabSize: Int,
@Since("1.6.0") @transient protected val sqlContext: SQLContext)
- extends Model[LDAModel] with LDAParams with Logging {
+ extends Model[LDAModel] with LDAParams with Logging with MLWritable {
// NOTE to developers:
// This abstraction should contain all important functionality for basic
LDA usage.
@@ -486,6 +487,64 @@ class LocalLDAModel private[ml] (
@Since("1.6.0")
override def isDistributed: Boolean = false
+
+ @Since("1.6.0")
+ override def write: MLWriter = new LocalLDAModel.LocalLDAModelWriter(this)
+}
+
+
+@Since("1.6.0")
+object LocalLDAModel extends MLReadable[LocalLDAModel] {
+
+ private[LocalLDAModel]
+ class LocalLDAModelWriter(instance: LocalLDAModel) extends MLWriter {
+
+private case class Data(
+vocabSize: Int,
+topicsMatrix: Matrix,
+docConcentration: Vector,
+topicConcentration: Double,
+gammaShape: Double)
+
+override protected def saveImpl(path: String): Unit = {
+ DefaultParamsWriter.saveMetadata(instance, path, sc)
+ val oldModel = instance.oldLocalModel
+ val data = Data(instance.vocabSize, oldModel.topicsMatrix,
oldModel.docConcentration,
+oldModel.topicConcentration, oldModel.gammaShape)
+ val dataPath = new Path(path, "data").toString
+
sqlContext.createDataFrame(Seq(data)).repartition(1).write.parquet(dataPath)
+}
+ }
+
+ private class LocalLDAModelReader extends MLReader[LocalLDAModel] {
+
+private val className = classOf[LocalLDAModel].getName
+
+override def load(path: String): LocalLDAModel = {
+ val metadata = DefaultParamsReader.loadMetadata(path, sc, className)
+ val dataPath = new Path(path, "data").toString
+ val data = sqlContext.read.parquet(dataPath)
+.select("vocabSize", "topicsMatrix", "docConcentration",
"topicConcentration",
+ "gammaShape")
+.head()
+ val vocabSize = data.getAs[Int](0)
+ val topicsMatrix = data.getAs[Matrix](1)
+ val docConcentration = data.getAs[Vector](2)
+ val topicConcentration = data.getAs[Double](3)
+ val gammaShape = data.getAs[Double](4)
+ val oldModel = new OldLocalLDAModel(topicsMatrix, docConcentration,
topicConcentratio
spark git commit: [SPARK-11847][ML] Model export/import for spark.ml: LDA
Repository: spark
Updated Branches:
refs/heads/master 9e24ba667 -> 52bc25c8e
[SPARK-11847][ML] Model export/import for spark.ml: LDA
Add read/write support to LDA, similar to ALS.
save/load for ml.LocalLDAModel is done.
For DistributedLDAModel, I'm not sure if we can invoke save on the
mllib.DistributedLDAModel directly. I'll send update after some test.
Author: Yuhao Yang
Closes #9894 from hhbyyh/ldaMLsave.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/52bc25c8
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/52bc25c8
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/52bc25c8
Branch: refs/heads/master
Commit: 52bc25c8e26d4be250d8ff7864067528f4f98592
Parents: 9e24ba6
Author: Yuhao Yang
Authored: Tue Nov 24 09:56:17 2015 -0800
Committer: Xiangrui Meng
Committed: Tue Nov 24 09:56:17 2015 -0800
--
.../org/apache/spark/ml/clustering/LDA.scala| 110 ++-
.../spark/mllib/clustering/LDAModel.scala | 4 +-
.../apache/spark/ml/clustering/LDASuite.scala | 44 +++-
3 files changed, 150 insertions(+), 8 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/52bc25c8/mllib/src/main/scala/org/apache/spark/ml/clustering/LDA.scala
--
diff --git a/mllib/src/main/scala/org/apache/spark/ml/clustering/LDA.scala
b/mllib/src/main/scala/org/apache/spark/ml/clustering/LDA.scala
index 92e0581..830510b 100644
--- a/mllib/src/main/scala/org/apache/spark/ml/clustering/LDA.scala
+++ b/mllib/src/main/scala/org/apache/spark/ml/clustering/LDA.scala
@@ -17,12 +17,13 @@
package org.apache.spark.ml.clustering
+import org.apache.hadoop.fs.Path
import org.apache.spark.Logging
import org.apache.spark.annotation.{Experimental, Since}
-import org.apache.spark.ml.util.{SchemaUtils, Identifiable}
import org.apache.spark.ml.{Estimator, Model}
import org.apache.spark.ml.param.shared.{HasCheckpointInterval,
HasFeaturesCol, HasSeed, HasMaxIter}
import org.apache.spark.ml.param._
+import org.apache.spark.ml.util._
import org.apache.spark.mllib.clustering.{DistributedLDAModel =>
OldDistributedLDAModel,
EMLDAOptimizer => OldEMLDAOptimizer, LDA => OldLDA, LDAModel =>
OldLDAModel,
LDAOptimizer => OldLDAOptimizer, LocalLDAModel => OldLocalLDAModel,
@@ -322,7 +323,7 @@ sealed abstract class LDAModel private[ml] (
@Since("1.6.0") override val uid: String,
@Since("1.6.0") val vocabSize: Int,
@Since("1.6.0") @transient protected val sqlContext: SQLContext)
- extends Model[LDAModel] with LDAParams with Logging {
+ extends Model[LDAModel] with LDAParams with Logging with MLWritable {
// NOTE to developers:
// This abstraction should contain all important functionality for basic
LDA usage.
@@ -486,6 +487,64 @@ class LocalLDAModel private[ml] (
@Since("1.6.0")
override def isDistributed: Boolean = false
+
+ @Since("1.6.0")
+ override def write: MLWriter = new LocalLDAModel.LocalLDAModelWriter(this)
+}
+
+
+@Since("1.6.0")
+object LocalLDAModel extends MLReadable[LocalLDAModel] {
+
+ private[LocalLDAModel]
+ class LocalLDAModelWriter(instance: LocalLDAModel) extends MLWriter {
+
+private case class Data(
+vocabSize: Int,
+topicsMatrix: Matrix,
+docConcentration: Vector,
+topicConcentration: Double,
+gammaShape: Double)
+
+override protected def saveImpl(path: String): Unit = {
+ DefaultParamsWriter.saveMetadata(instance, path, sc)
+ val oldModel = instance.oldLocalModel
+ val data = Data(instance.vocabSize, oldModel.topicsMatrix,
oldModel.docConcentration,
+oldModel.topicConcentration, oldModel.gammaShape)
+ val dataPath = new Path(path, "data").toString
+
sqlContext.createDataFrame(Seq(data)).repartition(1).write.parquet(dataPath)
+}
+ }
+
+ private class LocalLDAModelReader extends MLReader[LocalLDAModel] {
+
+private val className = classOf[LocalLDAModel].getName
+
+override def load(path: String): LocalLDAModel = {
+ val metadata = DefaultParamsReader.loadMetadata(path, sc, className)
+ val dataPath = new Path(path, "data").toString
+ val data = sqlContext.read.parquet(dataPath)
+.select("vocabSize", "topicsMatrix", "docConcentration",
"topicConcentration",
+ "gammaShape")
+.head()
+ val vocabSize = data.getAs[Int](0)
+ val topicsMatrix = data.getAs[Matrix](1)
+ val docConcentration = data.getAs[Vector](2)
+ val topicConcentration = data.getAs[Double](3)
+ val gammaShape = data.getAs[Double](4)
+ val oldModel = new OldLocalLDAModel(topicsMatrix, docConcentration,
topicConcentration,
+gammaShape)
+ val model = new LocalLDAModel(metadata.uid, vocabSize, oldModel,
sqlContext)
spark git commit: [SPARK-11926][SQL] unify GetStructField and GetInternalRowField
Repository: spark
Updated Branches:
refs/heads/master 52bc25c8e -> 19530da69
[SPARK-11926][SQL] unify GetStructField and GetInternalRowField
Author: Wenchen Fan
Closes #9909 from cloud-fan/get-struct.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/19530da6
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/19530da6
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/19530da6
Branch: refs/heads/master
Commit: 19530da6903fa59b051eec69b9c17e231c68454b
Parents: 52bc25c
Author: Wenchen Fan
Authored: Tue Nov 24 11:09:01 2015 -0800
Committer: Michael Armbrust
Committed: Tue Nov 24 11:09:01 2015 -0800
--
.../spark/sql/catalyst/ScalaReflection.scala| 2 +-
.../sql/catalyst/analysis/unresolved.scala | 8
.../catalyst/encoders/ExpressionEncoder.scala | 2 +-
.../sql/catalyst/encoders/RowEncoder.scala | 2 +-
.../sql/catalyst/expressions/Expression.scala | 2 +-
.../expressions/complexTypeExtractors.scala | 18 -
.../catalyst/expressions/namedExpressions.scala | 4 ++--
.../sql/catalyst/expressions/objects.scala | 21
.../catalyst/expressions/ComplexTypeSuite.scala | 4 ++--
9 files changed, 21 insertions(+), 42 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/19530da6/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 476bece..d133ad3 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
@@ -130,7 +130,7 @@ object ScalaReflection extends ScalaReflection {
/** Returns the current path with a field at ordinal extracted. */
def addToPathOrdinal(ordinal: Int, dataType: DataType): Expression = path
- .map(p => GetInternalRowField(p, ordinal, dataType))
+ .map(p => GetStructField(p, ordinal))
.getOrElse(BoundReference(ordinal, dataType, false))
/** Returns the current path or `BoundReference`. */
http://git-wip-us.apache.org/repos/asf/spark/blob/19530da6/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala
--
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala
index 6485bdf..1b2a8dc 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala
@@ -201,12 +201,12 @@ case class UnresolvedStar(target: Option[Seq[String]])
extends Star with Unevalu
if (attribute.isDefined) {
// This target resolved to an attribute in child. It must be a struct.
Expand it.
attribute.get.dataType match {
-case s: StructType => {
- s.fields.map( f => {
-val extract = GetStructField(attribute.get, f,
s.getFieldIndex(f.name).get)
+case s: StructType => s.zipWithIndex.map {
+ case (f, i) =>
+val extract = GetStructField(attribute.get, i)
Alias(extract, target.get + "." + f.name)()
- })
}
+
case _ => {
throw new AnalysisException("Can only star expand struct data types.
Attribute: `" +
target.get + "`")
http://git-wip-us.apache.org/repos/asf/spark/blob/19530da6/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/encoders/ExpressionEncoder.scala
--
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/encoders/ExpressionEncoder.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/encoders/ExpressionEncoder.scala
index 7bc9aed..0c10a56 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/encoders/ExpressionEncoder.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/encoders/ExpressionEncoder.scala
@@ -111,7 +111,7 @@ object ExpressionEncoder {
case UnresolvedAttribute(nameParts) =>
assert(nameParts.length == 1)
UnresolvedExtractValue(input, Literal(nameParts.head))
- case BoundReference(ordinal, dt, _) => GetInternalRowField(input,
ordinal, dt)
+ case BoundReference(ordinal, dt, _) => GetStructField(input, ordinal)
}
}
}
http://git-wip-us.apache.org/repos/asf/spark/blob/1953
spark git commit: [SPARK-11926][SQL] unify GetStructField and GetInternalRowField
Repository: spark
Updated Branches:
refs/heads/branch-1.6 af86c38db -> 927070d6d
[SPARK-11926][SQL] unify GetStructField and GetInternalRowField
Author: Wenchen Fan
Closes #9909 from cloud-fan/get-struct.
(cherry picked from commit 19530da6903fa59b051eec69b9c17e231c68454b)
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/927070d6
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/927070d6
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/927070d6
Branch: refs/heads/branch-1.6
Commit: 927070d6d75abcaac6d9676b1b9b556f21ec8536
Parents: af86c38
Author: Wenchen Fan
Authored: Tue Nov 24 11:09:01 2015 -0800
Committer: Michael Armbrust
Committed: Tue Nov 24 11:10:58 2015 -0800
--
.../spark/sql/catalyst/ScalaReflection.scala| 2 +-
.../sql/catalyst/analysis/unresolved.scala | 8
.../catalyst/encoders/ExpressionEncoder.scala | 2 +-
.../sql/catalyst/encoders/RowEncoder.scala | 2 +-
.../sql/catalyst/expressions/Expression.scala | 2 +-
.../expressions/complexTypeExtractors.scala | 18 -
.../catalyst/expressions/namedExpressions.scala | 4 ++--
.../sql/catalyst/expressions/objects.scala | 21
.../catalyst/expressions/ComplexTypeSuite.scala | 4 ++--
9 files changed, 21 insertions(+), 42 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/927070d6/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 476bece..d133ad3 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
@@ -130,7 +130,7 @@ object ScalaReflection extends ScalaReflection {
/** Returns the current path with a field at ordinal extracted. */
def addToPathOrdinal(ordinal: Int, dataType: DataType): Expression = path
- .map(p => GetInternalRowField(p, ordinal, dataType))
+ .map(p => GetStructField(p, ordinal))
.getOrElse(BoundReference(ordinal, dataType, false))
/** Returns the current path or `BoundReference`. */
http://git-wip-us.apache.org/repos/asf/spark/blob/927070d6/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala
--
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala
index 6485bdf..1b2a8dc 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala
@@ -201,12 +201,12 @@ case class UnresolvedStar(target: Option[Seq[String]])
extends Star with Unevalu
if (attribute.isDefined) {
// This target resolved to an attribute in child. It must be a struct.
Expand it.
attribute.get.dataType match {
-case s: StructType => {
- s.fields.map( f => {
-val extract = GetStructField(attribute.get, f,
s.getFieldIndex(f.name).get)
+case s: StructType => s.zipWithIndex.map {
+ case (f, i) =>
+val extract = GetStructField(attribute.get, i)
Alias(extract, target.get + "." + f.name)()
- })
}
+
case _ => {
throw new AnalysisException("Can only star expand struct data types.
Attribute: `" +
target.get + "`")
http://git-wip-us.apache.org/repos/asf/spark/blob/927070d6/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/encoders/ExpressionEncoder.scala
--
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/encoders/ExpressionEncoder.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/encoders/ExpressionEncoder.scala
index 7bc9aed..0c10a56 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/encoders/ExpressionEncoder.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/encoders/ExpressionEncoder.scala
@@ -111,7 +111,7 @@ object ExpressionEncoder {
case UnresolvedAttribute(nameParts) =>
assert(nameParts.length == 1)
UnresolvedExtractValue(input, Literal(nameParts.head))
- case BoundReference(ordinal, dt, _) => GetInternalRowField(input,
ordinal, dt)
+ case BoundReference(ordinal, dt, _) => Ge
spark git commit: [SPARK-11872] Prevent the call to SparkContext#stop() in the listener bus's thread
Repository: spark
Updated Branches:
refs/heads/master 19530da69 -> 81012546e
[SPARK-11872] Prevent the call to SparkContext#stop() in the listener bus's
thread
This is continuation of SPARK-11761
Andrew suggested adding this protection. See tail of
https://github.com/apache/spark/pull/9741
Author: tedyu
Closes #9852 from tedyu/master.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/81012546
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/81012546
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/81012546
Branch: refs/heads/master
Commit: 81012546ee5a80d2576740af0dad067b0f5962c5
Parents: 19530da
Author: tedyu
Authored: Tue Nov 24 12:22:33 2015 -0800
Committer: Shixiong Zhu
Committed: Tue Nov 24 12:22:33 2015 -0800
--
.../scala/org/apache/spark/SparkContext.scala | 4 +++
.../spark/scheduler/SparkListenerSuite.scala| 31
2 files changed, 35 insertions(+)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/81012546/core/src/main/scala/org/apache/spark/SparkContext.scala
--
diff --git a/core/src/main/scala/org/apache/spark/SparkContext.scala
b/core/src/main/scala/org/apache/spark/SparkContext.scala
index b153a7b..e19ba11 100644
--- a/core/src/main/scala/org/apache/spark/SparkContext.scala
+++ b/core/src/main/scala/org/apache/spark/SparkContext.scala
@@ -1694,6 +1694,10 @@ class SparkContext(config: SparkConf) extends Logging
with ExecutorAllocationCli
// Shut down the SparkContext.
def stop() {
+if (AsynchronousListenerBus.withinListenerThread.value) {
+ throw new SparkException("Cannot stop SparkContext within listener
thread of" +
+" AsynchronousListenerBus")
+}
// Use the stopping variable to ensure no contention for the stop scenario.
// Still track the stopped variable for use elsewhere in the code.
if (!stopped.compareAndSet(false, true)) {
http://git-wip-us.apache.org/repos/asf/spark/blob/81012546/core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala
--
diff --git
a/core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala
b/core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala
index 84e5458..f20d5be 100644
--- a/core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala
@@ -24,6 +24,7 @@ import scala.collection.JavaConverters._
import org.scalatest.Matchers
+import org.apache.spark.SparkException
import org.apache.spark.executor.TaskMetrics
import org.apache.spark.util.ResetSystemProperties
import org.apache.spark.{LocalSparkContext, SparkConf, SparkContext,
SparkFunSuite}
@@ -36,6 +37,21 @@ class SparkListenerSuite extends SparkFunSuite with
LocalSparkContext with Match
val jobCompletionTime = 1421191296660L
+ test("don't call sc.stop in listener") {
+sc = new SparkContext("local", "SparkListenerSuite")
+val listener = new SparkContextStoppingListener(sc)
+val bus = new LiveListenerBus
+bus.addListener(listener)
+
+// Starting listener bus should flush all buffered events
+bus.start(sc)
+bus.post(SparkListenerJobEnd(0, jobCompletionTime, JobSucceeded))
+bus.waitUntilEmpty(WAIT_TIMEOUT_MILLIS)
+
+bus.stop()
+assert(listener.sparkExSeen)
+ }
+
test("basic creation and shutdown of LiveListenerBus") {
val counter = new BasicJobCounter
val bus = new LiveListenerBus
@@ -443,6 +459,21 @@ private class BasicJobCounter extends SparkListener {
override def onJobEnd(job: SparkListenerJobEnd): Unit = count += 1
}
+/**
+ * A simple listener that tries to stop SparkContext.
+ */
+private class SparkContextStoppingListener(val sc: SparkContext) extends
SparkListener {
+ @volatile var sparkExSeen = false
+ override def onJobEnd(job: SparkListenerJobEnd): Unit = {
+try {
+ sc.stop()
+} catch {
+ case se: SparkException =>
+sparkExSeen = true
+}
+ }
+}
+
private class ListenerThatAcceptsSparkConf(conf: SparkConf) extends
SparkListener {
var count = 0
override def onJobEnd(job: SparkListenerJobEnd): Unit = count += 1
-
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
spark git commit: [SPARK-11872] Prevent the call to SparkContext#stop() in the listener bus's thread
Repository: spark
Updated Branches:
refs/heads/branch-1.6 927070d6d -> 0419fd361
[SPARK-11872] Prevent the call to SparkContext#stop() in the listener bus's
thread
This is continuation of SPARK-11761
Andrew suggested adding this protection. See tail of
https://github.com/apache/spark/pull/9741
Author: tedyu
Closes #9852 from tedyu/master.
(cherry picked from commit 81012546ee5a80d2576740af0dad067b0f5962c5)
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/0419fd36
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/0419fd36
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/0419fd36
Branch: refs/heads/branch-1.6
Commit: 0419fd3615ec73be1b2774e768238bf3f0cb4403
Parents: 927070d
Author: tedyu
Authored: Tue Nov 24 12:22:33 2015 -0800
Committer: Shixiong Zhu
Committed: Tue Nov 24 12:23:51 2015 -0800
--
.../scala/org/apache/spark/SparkContext.scala | 4 +++
.../spark/scheduler/SparkListenerSuite.scala| 31
2 files changed, 35 insertions(+)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/0419fd36/core/src/main/scala/org/apache/spark/SparkContext.scala
--
diff --git a/core/src/main/scala/org/apache/spark/SparkContext.scala
b/core/src/main/scala/org/apache/spark/SparkContext.scala
index af4456c..90480e5 100644
--- a/core/src/main/scala/org/apache/spark/SparkContext.scala
+++ b/core/src/main/scala/org/apache/spark/SparkContext.scala
@@ -1694,6 +1694,10 @@ class SparkContext(config: SparkConf) extends Logging
with ExecutorAllocationCli
// Shut down the SparkContext.
def stop() {
+if (AsynchronousListenerBus.withinListenerThread.value) {
+ throw new SparkException("Cannot stop SparkContext within listener
thread of" +
+" AsynchronousListenerBus")
+}
// Use the stopping variable to ensure no contention for the stop scenario.
// Still track the stopped variable for use elsewhere in the code.
if (!stopped.compareAndSet(false, true)) {
http://git-wip-us.apache.org/repos/asf/spark/blob/0419fd36/core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala
--
diff --git
a/core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala
b/core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala
index 84e5458..f20d5be 100644
--- a/core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala
@@ -24,6 +24,7 @@ import scala.collection.JavaConverters._
import org.scalatest.Matchers
+import org.apache.spark.SparkException
import org.apache.spark.executor.TaskMetrics
import org.apache.spark.util.ResetSystemProperties
import org.apache.spark.{LocalSparkContext, SparkConf, SparkContext,
SparkFunSuite}
@@ -36,6 +37,21 @@ class SparkListenerSuite extends SparkFunSuite with
LocalSparkContext with Match
val jobCompletionTime = 1421191296660L
+ test("don't call sc.stop in listener") {
+sc = new SparkContext("local", "SparkListenerSuite")
+val listener = new SparkContextStoppingListener(sc)
+val bus = new LiveListenerBus
+bus.addListener(listener)
+
+// Starting listener bus should flush all buffered events
+bus.start(sc)
+bus.post(SparkListenerJobEnd(0, jobCompletionTime, JobSucceeded))
+bus.waitUntilEmpty(WAIT_TIMEOUT_MILLIS)
+
+bus.stop()
+assert(listener.sparkExSeen)
+ }
+
test("basic creation and shutdown of LiveListenerBus") {
val counter = new BasicJobCounter
val bus = new LiveListenerBus
@@ -443,6 +459,21 @@ private class BasicJobCounter extends SparkListener {
override def onJobEnd(job: SparkListenerJobEnd): Unit = count += 1
}
+/**
+ * A simple listener that tries to stop SparkContext.
+ */
+private class SparkContextStoppingListener(val sc: SparkContext) extends
SparkListener {
+ @volatile var sparkExSeen = false
+ override def onJobEnd(job: SparkListenerJobEnd): Unit = {
+try {
+ sc.stop()
+} catch {
+ case se: SparkException =>
+sparkExSeen = true
+}
+ }
+}
+
private class ListenerThatAcceptsSparkConf(conf: SparkConf) extends
SparkListener {
var count = 0
override def onJobEnd(job: SparkListenerJobEnd): Unit = count += 1
-
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
spark git commit: [SPARK-11946][SQL] Audit pivot API for 1.6.
Repository: spark
Updated Branches:
refs/heads/master 81012546e -> f31527227
[SPARK-11946][SQL] Audit pivot API for 1.6.
Currently pivot's signature looks like
```scala
scala.annotation.varargs
def pivot(pivotColumn: Column, values: Column*): GroupedData
scala.annotation.varargs
def pivot(pivotColumn: String, values: Any*): GroupedData
```
I think we can remove the one that takes "Column" types, since callers should
always be passing in literals. It'd also be more clear if the values are not
varargs, but rather Seq or java.util.List.
I also made similar changes for Python.
Author: Reynold Xin
Closes #9929 from rxin/SPARK-11946.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/f3152722
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/f3152722
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/f3152722
Branch: refs/heads/master
Commit: f3152722791b163fa66597b3684009058195ba33
Parents: 8101254
Author: Reynold Xin
Authored: Tue Nov 24 12:54:37 2015 -0800
Committer: Reynold Xin
Committed: Tue Nov 24 12:54:37 2015 -0800
--
.../apache/spark/scheduler/DAGScheduler.scala | 1 -
python/pyspark/sql/group.py | 12 +-
.../sql/catalyst/expressions/literals.scala | 1 +
.../org/apache/spark/sql/GroupedData.scala | 154 +++
.../apache/spark/sql/JavaDataFrameSuite.java| 16 ++
.../apache/spark/sql/DataFramePivotSuite.scala | 21 +--
.../org/apache/spark/sql/test/SQLTestData.scala | 1 +
7 files changed, 125 insertions(+), 81 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/f3152722/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
--
diff --git a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
index ae725b4..77a184d 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
@@ -1574,7 +1574,6 @@ class DAGScheduler(
}
def stop() {
-logInfo("Stopping DAGScheduler")
messageScheduler.shutdownNow()
eventProcessLoop.stop()
taskScheduler.stop()
http://git-wip-us.apache.org/repos/asf/spark/blob/f3152722/python/pyspark/sql/group.py
--
diff --git a/python/pyspark/sql/group.py b/python/pyspark/sql/group.py
index 227f40b..d8ed7eb 100644
--- a/python/pyspark/sql/group.py
+++ b/python/pyspark/sql/group.py
@@ -168,20 +168,24 @@ class GroupedData(object):
"""
@since(1.6)
-def pivot(self, pivot_col, *values):
+def pivot(self, pivot_col, values=None):
"""Pivots a column of the current DataFrame and preform the specified
aggregation.
:param pivot_col: Column to pivot
:param values: Optional list of values of pivotColumn that will be
translated to columns in
the output data frame. If values are not provided the method with
do an immediate call
to .distinct() on the pivot column.
->>> df4.groupBy("year").pivot("course", "dotNET",
"Java").sum("earnings").collect()
+
+>>> df4.groupBy("year").pivot("course", ["dotNET",
"Java"]).sum("earnings").collect()
[Row(year=2012, dotNET=15000, Java=2), Row(year=2013,
dotNET=48000, Java=3)]
+
>>> df4.groupBy("year").pivot("course").sum("earnings").collect()
[Row(year=2012, Java=2, dotNET=15000), Row(year=2013, Java=3,
dotNET=48000)]
"""
-jgd = self._jdf.pivot(_to_java_column(pivot_col),
- _to_seq(self.sql_ctx._sc, values,
_create_column_from_literal))
+if values is None:
+jgd = self._jdf.pivot(pivot_col)
+else:
+jgd = self._jdf.pivot(pivot_col, values)
return GroupedData(jgd, self.sql_ctx)
http://git-wip-us.apache.org/repos/asf/spark/blob/f3152722/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/literals.scala
--
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/literals.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/literals.scala
index e34fd49..68ec688 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/literals.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/literals.scala
@@ -44,6 +44,7 @@ object Literal {
case a: Array[Byte] => Literal(a, BinaryType)
case i: CalendarInterval => Literal(i, CalendarIntervalType)
case null => Literal(null, NullType)
+case v: Lite
spark git commit: [SPARK-11946][SQL] Audit pivot API for 1.6.
Repository: spark
Updated Branches:
refs/heads/branch-1.6 0419fd361 -> 3f40af574
[SPARK-11946][SQL] Audit pivot API for 1.6.
Currently pivot's signature looks like
```scala
scala.annotation.varargs
def pivot(pivotColumn: Column, values: Column*): GroupedData
scala.annotation.varargs
def pivot(pivotColumn: String, values: Any*): GroupedData
```
I think we can remove the one that takes "Column" types, since callers should
always be passing in literals. It'd also be more clear if the values are not
varargs, but rather Seq or java.util.List.
I also made similar changes for Python.
Author: Reynold Xin
Closes #9929 from rxin/SPARK-11946.
(cherry picked from commit f3152722791b163fa66597b3684009058195ba33)
Signed-off-by: Reynold Xin
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/3f40af57
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/3f40af57
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/3f40af57
Branch: refs/heads/branch-1.6
Commit: 3f40af574ed14909d30c10cad24d747fba49b783
Parents: 0419fd3
Author: Reynold Xin
Authored: Tue Nov 24 12:54:37 2015 -0800
Committer: Reynold Xin
Committed: Tue Nov 24 12:54:44 2015 -0800
--
.../apache/spark/scheduler/DAGScheduler.scala | 1 -
python/pyspark/sql/group.py | 12 +-
.../sql/catalyst/expressions/literals.scala | 1 +
.../org/apache/spark/sql/GroupedData.scala | 154 +++
.../apache/spark/sql/JavaDataFrameSuite.java| 16 ++
.../apache/spark/sql/DataFramePivotSuite.scala | 21 +--
.../org/apache/spark/sql/test/SQLTestData.scala | 1 +
7 files changed, 125 insertions(+), 81 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/3f40af57/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
--
diff --git a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
index ae725b4..77a184d 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
@@ -1574,7 +1574,6 @@ class DAGScheduler(
}
def stop() {
-logInfo("Stopping DAGScheduler")
messageScheduler.shutdownNow()
eventProcessLoop.stop()
taskScheduler.stop()
http://git-wip-us.apache.org/repos/asf/spark/blob/3f40af57/python/pyspark/sql/group.py
--
diff --git a/python/pyspark/sql/group.py b/python/pyspark/sql/group.py
index 227f40b..d8ed7eb 100644
--- a/python/pyspark/sql/group.py
+++ b/python/pyspark/sql/group.py
@@ -168,20 +168,24 @@ class GroupedData(object):
"""
@since(1.6)
-def pivot(self, pivot_col, *values):
+def pivot(self, pivot_col, values=None):
"""Pivots a column of the current DataFrame and preform the specified
aggregation.
:param pivot_col: Column to pivot
:param values: Optional list of values of pivotColumn that will be
translated to columns in
the output data frame. If values are not provided the method with
do an immediate call
to .distinct() on the pivot column.
->>> df4.groupBy("year").pivot("course", "dotNET",
"Java").sum("earnings").collect()
+
+>>> df4.groupBy("year").pivot("course", ["dotNET",
"Java"]).sum("earnings").collect()
[Row(year=2012, dotNET=15000, Java=2), Row(year=2013,
dotNET=48000, Java=3)]
+
>>> df4.groupBy("year").pivot("course").sum("earnings").collect()
[Row(year=2012, Java=2, dotNET=15000), Row(year=2013, Java=3,
dotNET=48000)]
"""
-jgd = self._jdf.pivot(_to_java_column(pivot_col),
- _to_seq(self.sql_ctx._sc, values,
_create_column_from_literal))
+if values is None:
+jgd = self._jdf.pivot(pivot_col)
+else:
+jgd = self._jdf.pivot(pivot_col, values)
return GroupedData(jgd, self.sql_ctx)
http://git-wip-us.apache.org/repos/asf/spark/blob/3f40af57/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/literals.scala
--
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/literals.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/literals.scala
index e34fd49..68ec688 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/literals.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/literals.scala
@@ -44,6 +44,7 @@ object Literal {
case a: Array[Byte] => Literal(a, BinaryType)
case i: Calend
spark git commit: [SPARK-11929][CORE] Make the repl log4j configuration override the root logger.
Repository: spark
Updated Branches:
refs/heads/master f31527227 -> e6dd23746
[SPARK-11929][CORE] Make the repl log4j configuration override the root logger.
In the default Spark distribution, there are currently two separate
log4j config files, with different default values for the root logger,
so that when running the shell you have a different default log level.
This makes the shell more usable, since the logs don't overwhelm the
output.
But if you install a custom log4j.properties, you lose that, because
then it's going to be used no matter whether you're running a regular
app or the shell.
With this change, the overriding of the log level is done differently;
the log level repl's main class (org.apache.spark.repl.Main) is used
to define the root logger's level when running the shell, defaulting
to WARN if it's not set explicitly.
On a somewhat related change, the shell output about the "sc" variable
was changed a bit to contain a little more useful information about
the application, since when the root logger's log level is WARN, that
information is never shown to the user.
Author: Marcelo Vanzin
Closes #9816 from vanzin/shell-logging.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/e6dd2374
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/e6dd2374
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/e6dd2374
Branch: refs/heads/master
Commit: e6dd237463d2de8c506f0735dfdb3f43e8122513
Parents: f315272
Author: Marcelo Vanzin
Authored: Tue Nov 24 15:08:02 2015 -0600
Committer: Imran Rashid
Committed: Tue Nov 24 15:08:02 2015 -0600
--
conf/log4j.properties.template | 5 +++
.../apache/spark/log4j-defaults-repl.properties | 33 --
.../org/apache/spark/log4j-defaults.properties | 5 +++
.../main/scala/org/apache/spark/Logging.scala | 45 ++--
.../org/apache/spark/repl/SparkILoopInit.scala | 21 -
.../org/apache/spark/repl/SparkILoop.scala | 25 +--
6 files changed, 57 insertions(+), 77 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/e6dd2374/conf/log4j.properties.template
--
diff --git a/conf/log4j.properties.template b/conf/log4j.properties.template
index f3046be..9809b0c 100644
--- a/conf/log4j.properties.template
+++ b/conf/log4j.properties.template
@@ -22,6 +22,11 @@ log4j.appender.console.target=System.err
log4j.appender.console.layout=org.apache.log4j.PatternLayout
log4j.appender.console.layout.ConversionPattern=%d{yy/MM/dd HH:mm:ss} %p
%c{1}: %m%n
+# Set the default spark-shell log level to WARN. When running the spark-shell,
the
+# log level for this class is used to overwrite the root logger's log level,
so that
+# the user can have different defaults for the shell and regular Spark apps.
+log4j.logger.org.apache.spark.repl.Main=WARN
+
# Settings to quiet third party logs that are too verbose
log4j.logger.org.spark-project.jetty=WARN
log4j.logger.org.spark-project.jetty.util.component.AbstractLifeCycle=ERROR
http://git-wip-us.apache.org/repos/asf/spark/blob/e6dd2374/core/src/main/resources/org/apache/spark/log4j-defaults-repl.properties
--
diff --git
a/core/src/main/resources/org/apache/spark/log4j-defaults-repl.properties
b/core/src/main/resources/org/apache/spark/log4j-defaults-repl.properties
deleted file mode 100644
index c85abc3..000
--- a/core/src/main/resources/org/apache/spark/log4j-defaults-repl.properties
+++ /dev/null
@@ -1,33 +0,0 @@
-#
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements. See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You under the Apache License, Version 2.0
-# (the "License"); you may not use this file except in compliance with
-# the License. You may obtain a copy of the License at
-#
-#http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-# Set everything to be logged to the console
-log4j.rootCategory=WARN, console
-log4j.appender.console=org.apache.log4j.ConsoleAppender
-log4j.appender.console.target=System.err
-log4j.appender.console.layout=org.apache.log4j.PatternLayout
-log4j.appender.console.layout.ConversionPattern=%d{yy/MM/dd HH:mm:ss} %p
%c{1}: %m%n
-
-# Settings to quiet third party logs that are too verbose
-log4j.logger.or
spark git commit: [SPARK-11805] free the array in UnsafeExternalSorter during spilling
Repository: spark
Updated Branches:
refs/heads/branch-1.6 3f40af574 -> 015569341
[SPARK-11805] free the array in UnsafeExternalSorter during spilling
After calling spill() on SortedIterator, the array inside InMemorySorter is not
needed, it should be freed during spilling, this could help to join multiple
tables with limited memory.
Author: Davies Liu
Closes #9793 from davies/free_array.
(cherry picked from commit 58d9b260556a89a3d0832d583acafba1df7c6751)
Signed-off-by: Josh Rosen
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/01556934
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/01556934
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/01556934
Branch: refs/heads/branch-1.6
Commit: 015569341dc8031b62ea94a2d194bbd35567f727
Parents: 3f40af5
Author: Davies Liu
Authored: Tue Nov 24 14:33:28 2015 -0800
Committer: Josh Rosen
Committed: Tue Nov 24 14:33:40 2015 -0800
--
.../unsafe/sort/UnsafeExternalSorter.java | 10 ---
.../unsafe/sort/UnsafeInMemorySorter.java | 31
2 files changed, 19 insertions(+), 22 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/01556934/core/src/main/java/org/apache/spark/util/collection/unsafe/sort/UnsafeExternalSorter.java
--
diff --git
a/core/src/main/java/org/apache/spark/util/collection/unsafe/sort/UnsafeExternalSorter.java
b/core/src/main/java/org/apache/spark/util/collection/unsafe/sort/UnsafeExternalSorter.java
index 9a7b2ad..2e40312 100644
---
a/core/src/main/java/org/apache/spark/util/collection/unsafe/sort/UnsafeExternalSorter.java
+++
b/core/src/main/java/org/apache/spark/util/collection/unsafe/sort/UnsafeExternalSorter.java
@@ -468,6 +468,12 @@ public final class UnsafeExternalSorter extends
MemoryConsumer {
}
allocatedPages.clear();
}
+
+// in-memory sorter will not be used after spilling
+assert(inMemSorter != null);
+released += inMemSorter.getMemoryUsage();
+inMemSorter.free();
+inMemSorter = null;
return released;
}
}
@@ -489,10 +495,6 @@ public final class UnsafeExternalSorter extends
MemoryConsumer {
}
upstream = nextUpstream;
nextUpstream = null;
-
- assert(inMemSorter != null);
- inMemSorter.free();
- inMemSorter = null;
}
numRecords--;
upstream.loadNext();
http://git-wip-us.apache.org/repos/asf/spark/blob/01556934/core/src/main/java/org/apache/spark/util/collection/unsafe/sort/UnsafeInMemorySorter.java
--
diff --git
a/core/src/main/java/org/apache/spark/util/collection/unsafe/sort/UnsafeInMemorySorter.java
b/core/src/main/java/org/apache/spark/util/collection/unsafe/sort/UnsafeInMemorySorter.java
index a218ad4..dce1f15 100644
---
a/core/src/main/java/org/apache/spark/util/collection/unsafe/sort/UnsafeInMemorySorter.java
+++
b/core/src/main/java/org/apache/spark/util/collection/unsafe/sort/UnsafeInMemorySorter.java
@@ -108,6 +108,7 @@ public final class UnsafeInMemorySorter {
*/
public void free() {
consumer.freeArray(array);
+array = null;
}
public void reset() {
@@ -160,28 +161,22 @@ public final class UnsafeInMemorySorter {
pos++;
}
- public static final class SortedIterator extends UnsafeSorterIterator {
+ public final class SortedIterator extends UnsafeSorterIterator {
-private final TaskMemoryManager memoryManager;
-private final int sortBufferInsertPosition;
-private final LongArray sortBuffer;
-private int position = 0;
+private final int numRecords;
+private int position;
private Object baseObject;
private long baseOffset;
private long keyPrefix;
private int recordLength;
-private SortedIterator(
-TaskMemoryManager memoryManager,
-int sortBufferInsertPosition,
-LongArray sortBuffer) {
- this.memoryManager = memoryManager;
- this.sortBufferInsertPosition = sortBufferInsertPosition;
- this.sortBuffer = sortBuffer;
+private SortedIterator(int numRecords) {
+ this.numRecords = numRecords;
+ this.position = 0;
}
public SortedIterator clone () {
- SortedIterator iter = new SortedIterator(memoryManager,
sortBufferInsertPosition, sortBuffer);
+ SortedIterator iter = new SortedIterator(numRecords);
iter.position = position;
iter.baseObject = baseObject;
iter.baseOffset = baseOffset;
@@ -192,21 +187,21 @@ public final class UnsafeInMemorySorter {
@Override
public boolean hasNext() {
- return position < sortBufferInsertPosition;
+ return position
spark git commit: [SPARK-11805] free the array in UnsafeExternalSorter during spilling
Repository: spark
Updated Branches:
refs/heads/master e6dd23746 -> 58d9b2605
[SPARK-11805] free the array in UnsafeExternalSorter during spilling
After calling spill() on SortedIterator, the array inside InMemorySorter is not
needed, it should be freed during spilling, this could help to join multiple
tables with limited memory.
Author: Davies Liu
Closes #9793 from davies/free_array.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/58d9b260
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/58d9b260
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/58d9b260
Branch: refs/heads/master
Commit: 58d9b260556a89a3d0832d583acafba1df7c6751
Parents: e6dd237
Author: Davies Liu
Authored: Tue Nov 24 14:33:28 2015 -0800
Committer: Josh Rosen
Committed: Tue Nov 24 14:33:28 2015 -0800
--
.../unsafe/sort/UnsafeExternalSorter.java | 10 ---
.../unsafe/sort/UnsafeInMemorySorter.java | 31
2 files changed, 19 insertions(+), 22 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/58d9b260/core/src/main/java/org/apache/spark/util/collection/unsafe/sort/UnsafeExternalSorter.java
--
diff --git
a/core/src/main/java/org/apache/spark/util/collection/unsafe/sort/UnsafeExternalSorter.java
b/core/src/main/java/org/apache/spark/util/collection/unsafe/sort/UnsafeExternalSorter.java
index 9a7b2ad..2e40312 100644
---
a/core/src/main/java/org/apache/spark/util/collection/unsafe/sort/UnsafeExternalSorter.java
+++
b/core/src/main/java/org/apache/spark/util/collection/unsafe/sort/UnsafeExternalSorter.java
@@ -468,6 +468,12 @@ public final class UnsafeExternalSorter extends
MemoryConsumer {
}
allocatedPages.clear();
}
+
+// in-memory sorter will not be used after spilling
+assert(inMemSorter != null);
+released += inMemSorter.getMemoryUsage();
+inMemSorter.free();
+inMemSorter = null;
return released;
}
}
@@ -489,10 +495,6 @@ public final class UnsafeExternalSorter extends
MemoryConsumer {
}
upstream = nextUpstream;
nextUpstream = null;
-
- assert(inMemSorter != null);
- inMemSorter.free();
- inMemSorter = null;
}
numRecords--;
upstream.loadNext();
http://git-wip-us.apache.org/repos/asf/spark/blob/58d9b260/core/src/main/java/org/apache/spark/util/collection/unsafe/sort/UnsafeInMemorySorter.java
--
diff --git
a/core/src/main/java/org/apache/spark/util/collection/unsafe/sort/UnsafeInMemorySorter.java
b/core/src/main/java/org/apache/spark/util/collection/unsafe/sort/UnsafeInMemorySorter.java
index a218ad4..dce1f15 100644
---
a/core/src/main/java/org/apache/spark/util/collection/unsafe/sort/UnsafeInMemorySorter.java
+++
b/core/src/main/java/org/apache/spark/util/collection/unsafe/sort/UnsafeInMemorySorter.java
@@ -108,6 +108,7 @@ public final class UnsafeInMemorySorter {
*/
public void free() {
consumer.freeArray(array);
+array = null;
}
public void reset() {
@@ -160,28 +161,22 @@ public final class UnsafeInMemorySorter {
pos++;
}
- public static final class SortedIterator extends UnsafeSorterIterator {
+ public final class SortedIterator extends UnsafeSorterIterator {
-private final TaskMemoryManager memoryManager;
-private final int sortBufferInsertPosition;
-private final LongArray sortBuffer;
-private int position = 0;
+private final int numRecords;
+private int position;
private Object baseObject;
private long baseOffset;
private long keyPrefix;
private int recordLength;
-private SortedIterator(
-TaskMemoryManager memoryManager,
-int sortBufferInsertPosition,
-LongArray sortBuffer) {
- this.memoryManager = memoryManager;
- this.sortBufferInsertPosition = sortBufferInsertPosition;
- this.sortBuffer = sortBuffer;
+private SortedIterator(int numRecords) {
+ this.numRecords = numRecords;
+ this.position = 0;
}
public SortedIterator clone () {
- SortedIterator iter = new SortedIterator(memoryManager,
sortBufferInsertPosition, sortBuffer);
+ SortedIterator iter = new SortedIterator(numRecords);
iter.position = position;
iter.baseObject = baseObject;
iter.baseOffset = baseOffset;
@@ -192,21 +187,21 @@ public final class UnsafeInMemorySorter {
@Override
public boolean hasNext() {
- return position < sortBufferInsertPosition;
+ return position / 2 < numRecords;
}
public int numRecordsLeft() {
- return (sortBufferInsertPosition - p
spark git commit: Added a line of comment to explain why the extra sort exists in pivot.
Repository: spark Updated Branches: refs/heads/master 58d9b2605 -> 34ca392da Added a line of comment to explain why the extra sort exists in pivot. Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/34ca392d Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/34ca392d Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/34ca392d Branch: refs/heads/master Commit: 34ca392da7097a1fbe48cd6c3ebff51453ca26ca Parents: 58d9b26 Author: Reynold Xin Authored: Tue Nov 24 14:51:01 2015 -0800 Committer: Reynold Xin Committed: Tue Nov 24 14:51:01 2015 -0800 -- sql/core/src/main/scala/org/apache/spark/sql/GroupedData.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) -- http://git-wip-us.apache.org/repos/asf/spark/blob/34ca392d/sql/core/src/main/scala/org/apache/spark/sql/GroupedData.scala -- diff --git a/sql/core/src/main/scala/org/apache/spark/sql/GroupedData.scala b/sql/core/src/main/scala/org/apache/spark/sql/GroupedData.scala index ee7150c..abd531c 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/GroupedData.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/GroupedData.scala @@ -304,7 +304,7 @@ class GroupedData protected[sql]( // Get the distinct values of the column and sort them so its consistent val values = df.select(pivotColumn) .distinct() - .sort(pivotColumn) + .sort(pivotColumn) // ensure that the output columns are in a consistent logical order .map(_.get(0)) .take(maxValues + 1) .toSeq - To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org
spark git commit: [SPARK-11783][SQL] Fixes execution Hive client when using remote Hive metastore
Repository: spark
Updated Branches:
refs/heads/master 34ca392da -> c7f95df5c
[SPARK-11783][SQL] Fixes execution Hive client when using remote Hive metastore
When using remote Hive metastore, `hive.metastore.uris` is set to the metastore
URI. However, it overrides `javax.jdo.option.ConnectionURL` unexpectedly, thus
the execution Hive client connects to the actual remote Hive metastore instead
of the Derby metastore created in the temporary directory. Cleaning this
configuration for the execution Hive client fixes this issue.
Author: Cheng Lian
Closes #9895 from liancheng/spark-11783.clean-remote-metastore-config.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/c7f95df5
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/c7f95df5
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/c7f95df5
Branch: refs/heads/master
Commit: c7f95df5c6d8eb2e6f11cf58b704fea34326a5f2
Parents: 34ca392
Author: Cheng Lian
Authored: Tue Nov 24 14:59:14 2015 -0800
Committer: Yin Huai
Committed: Tue Nov 24 15:08:53 2015 -0800
--
.../org/apache/spark/sql/hive/HiveContext.scala | 15 +++
1 file changed, 15 insertions(+)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/c7f95df5/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveContext.scala
--
diff --git
a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveContext.scala
b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveContext.scala
index c0bb5af..8a42641 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveContext.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveContext.scala
@@ -736,6 +736,21 @@ private[hive] object HiveContext {
s"jdbc:derby:;databaseName=${localMetastore.getAbsolutePath};create=true")
propMap.put("datanucleus.rdbms.datastoreAdapterClassName",
"org.datanucleus.store.rdbms.adapter.DerbyAdapter")
+
+// SPARK-11783: When "hive.metastore.uris" is set, the metastore
connection mode will be
+// remote
(https://cwiki.apache.org/confluence/display/Hive/AdminManual+MetastoreAdmin
+// mentions that "If hive.metastore.uris is empty local mode is assumed,
remote otherwise").
+// Remote means that the metastore server is running in its own process.
+// When the mode is remote, configurations like
"javax.jdo.option.ConnectionURL" will not be
+// used (because they are used by remote metastore server that talks to
the database).
+// Because execution Hive should always connects to a embedded derby
metastore.
+// We have to remove the value of hive.metastore.uris. So, the execution
Hive client connects
+// to the actual embedded derby metastore instead of the remote metastore.
+// You can search HiveConf.ConfVars.METASTOREURIS in the code of HiveConf
(in Hive's repo).
+// Then, you will find that the local metastore mode is only set to true
when
+// hive.metastore.uris is not set.
+propMap.put(ConfVars.METASTOREURIS.varname, "")
+
propMap.toMap
}
-
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
spark git commit: [SPARK-11783][SQL] Fixes execution Hive client when using remote Hive metastore
Repository: spark
Updated Branches:
refs/heads/branch-1.6 015569341 -> 3f15ad783
[SPARK-11783][SQL] Fixes execution Hive client when using remote Hive metastore
When using remote Hive metastore, `hive.metastore.uris` is set to the metastore
URI. However, it overrides `javax.jdo.option.ConnectionURL` unexpectedly, thus
the execution Hive client connects to the actual remote Hive metastore instead
of the Derby metastore created in the temporary directory. Cleaning this
configuration for the execution Hive client fixes this issue.
Author: Cheng Lian
Closes #9895 from liancheng/spark-11783.clean-remote-metastore-config.
(cherry picked from commit c7f95df5c6d8eb2e6f11cf58b704fea34326a5f2)
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/3f15ad78
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/3f15ad78
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/3f15ad78
Branch: refs/heads/branch-1.6
Commit: 3f15ad783886d077b5ecf5eee1edfc2d699bb35a
Parents: 0155693
Author: Cheng Lian
Authored: Tue Nov 24 14:59:14 2015 -0800
Committer: Yin Huai
Committed: Tue Nov 24 15:09:17 2015 -0800
--
.../org/apache/spark/sql/hive/HiveContext.scala | 15 +++
1 file changed, 15 insertions(+)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/3f15ad78/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveContext.scala
--
diff --git
a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveContext.scala
b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveContext.scala
index c0bb5af..8a42641 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveContext.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveContext.scala
@@ -736,6 +736,21 @@ private[hive] object HiveContext {
s"jdbc:derby:;databaseName=${localMetastore.getAbsolutePath};create=true")
propMap.put("datanucleus.rdbms.datastoreAdapterClassName",
"org.datanucleus.store.rdbms.adapter.DerbyAdapter")
+
+// SPARK-11783: When "hive.metastore.uris" is set, the metastore
connection mode will be
+// remote
(https://cwiki.apache.org/confluence/display/Hive/AdminManual+MetastoreAdmin
+// mentions that "If hive.metastore.uris is empty local mode is assumed,
remote otherwise").
+// Remote means that the metastore server is running in its own process.
+// When the mode is remote, configurations like
"javax.jdo.option.ConnectionURL" will not be
+// used (because they are used by remote metastore server that talks to
the database).
+// Because execution Hive should always connects to a embedded derby
metastore.
+// We have to remove the value of hive.metastore.uris. So, the execution
Hive client connects
+// to the actual embedded derby metastore instead of the remote metastore.
+// You can search HiveConf.ConfVars.METASTOREURIS in the code of HiveConf
(in Hive's repo).
+// Then, you will find that the local metastore mode is only set to true
when
+// hive.metastore.uris is not set.
+propMap.put(ConfVars.METASTOREURIS.varname, "")
+
propMap.toMap
}
-
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
spark git commit: [SPARK-11914][SQL] Support coalesce and repartition in Dataset APIs
Repository: spark
Updated Branches:
refs/heads/branch-1.6 3f15ad783 -> 6d8c4c644
[SPARK-11914][SQL] Support coalesce and repartition in Dataset APIs
This PR is to provide two common `coalesce` and `repartition` in Dataset APIs.
After reading the comments of SPARK-, I am unclear about the plan for
supporting re-partitioning in Dataset APIs. Currently, both RDD APIs and
Dataframe APIs provide users such a flexibility to control the number of
partitions.
In most traditional RDBMS, they expose the number of partitions, the
partitioning columns, the table partitioning methods to DBAs for performance
tuning and storage planning. Normally, these parameters could largely affect
the query performance. Since the actual performance depends on the workload
types, I think it is almost impossible to automate the discovery of the best
partitioning strategy for all the scenarios.
I am wondering if Dataset APIs are planning to hide these APIs from users? Feel
free to reject my PR if it does not match the plan.
Thank you for your answers. marmbrus rxin cloud-fan
Author: gatorsmile
Closes #9899 from gatorsmile/coalesce.
(cherry picked from commit 238ae51b66ac12d15fba6aff061804004c5ca6cb)
Signed-off-by: Reynold Xin
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/6d8c4c64
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/6d8c4c64
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/6d8c4c64
Branch: refs/heads/branch-1.6
Commit: 6d8c4c6448e898dad36e243da3eec647d2a45076
Parents: 3f15ad7
Author: gatorsmile
Authored: Tue Nov 24 15:54:10 2015 -0800
Committer: Reynold Xin
Committed: Tue Nov 24 15:54:16 2015 -0800
--
.../scala/org/apache/spark/sql/Dataset.scala | 19 +++
.../org/apache/spark/sql/DatasetSuite.scala | 15 +++
2 files changed, 34 insertions(+)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/6d8c4c64/sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala
--
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala
b/sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala
index 0764750..17e2611 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala
@@ -152,6 +152,25 @@ class Dataset[T] private[sql](
*/
def count(): Long = toDF().count()
+ /**
+* Returns a new [[Dataset]] that has exactly `numPartitions` partitions.
+* @since 1.6.0
+*/
+ def repartition(numPartitions: Int): Dataset[T] = withPlan {
+Repartition(numPartitions, shuffle = true, _)
+ }
+
+ /**
+* Returns a new [[Dataset]] that has exactly `numPartitions` partitions.
+* Similar to coalesce defined on an [[RDD]], this operation results in a
narrow dependency, e.g.
+* if you go from 1000 partitions to 100 partitions, there will not be a
shuffle, instead each of
+* the 100 new partitions will claim 10 of the current partitions.
+* @since 1.6.0
+*/
+ def coalesce(numPartitions: Int): Dataset[T] = withPlan {
+Repartition(numPartitions, shuffle = false, _)
+ }
+
/* *** *
* Functional Operations *
* *** */
http://git-wip-us.apache.org/repos/asf/spark/blob/6d8c4c64/sql/core/src/test/scala/org/apache/spark/sql/DatasetSuite.scala
--
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/DatasetSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/DatasetSuite.scala
index 13eede1..c253fdb 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/DatasetSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/DatasetSuite.scala
@@ -52,6 +52,21 @@ class DatasetSuite extends QueryTest with SharedSQLContext {
assert(ds.takeAsList(1).get(0) == item)
}
+ test("coalesce, repartition") {
+val data = (1 to 100).map(i => ClassData(i.toString, i))
+val ds = data.toDS()
+
+assert(ds.repartition(10).rdd.partitions.length == 10)
+checkAnswer(
+ ds.repartition(10),
+ data: _*)
+
+assert(ds.coalesce(1).rdd.partitions.length == 1)
+checkAnswer(
+ ds.coalesce(1),
+ data: _*)
+ }
+
test("as tuple") {
val data = Seq(("a", 1), ("b", 2)).toDF("a", "b")
checkAnswer(
-
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
spark git commit: [SPARK-11914][SQL] Support coalesce and repartition in Dataset APIs
Repository: spark
Updated Branches:
refs/heads/master c7f95df5c -> 238ae51b6
[SPARK-11914][SQL] Support coalesce and repartition in Dataset APIs
This PR is to provide two common `coalesce` and `repartition` in Dataset APIs.
After reading the comments of SPARK-, I am unclear about the plan for
supporting re-partitioning in Dataset APIs. Currently, both RDD APIs and
Dataframe APIs provide users such a flexibility to control the number of
partitions.
In most traditional RDBMS, they expose the number of partitions, the
partitioning columns, the table partitioning methods to DBAs for performance
tuning and storage planning. Normally, these parameters could largely affect
the query performance. Since the actual performance depends on the workload
types, I think it is almost impossible to automate the discovery of the best
partitioning strategy for all the scenarios.
I am wondering if Dataset APIs are planning to hide these APIs from users? Feel
free to reject my PR if it does not match the plan.
Thank you for your answers. marmbrus rxin cloud-fan
Author: gatorsmile
Closes #9899 from gatorsmile/coalesce.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/238ae51b
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/238ae51b
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/238ae51b
Branch: refs/heads/master
Commit: 238ae51b66ac12d15fba6aff061804004c5ca6cb
Parents: c7f95df
Author: gatorsmile
Authored: Tue Nov 24 15:54:10 2015 -0800
Committer: Reynold Xin
Committed: Tue Nov 24 15:54:10 2015 -0800
--
.../scala/org/apache/spark/sql/Dataset.scala | 19 +++
.../org/apache/spark/sql/DatasetSuite.scala | 15 +++
2 files changed, 34 insertions(+)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/238ae51b/sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala
--
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala
b/sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala
index 0764750..17e2611 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala
@@ -152,6 +152,25 @@ class Dataset[T] private[sql](
*/
def count(): Long = toDF().count()
+ /**
+* Returns a new [[Dataset]] that has exactly `numPartitions` partitions.
+* @since 1.6.0
+*/
+ def repartition(numPartitions: Int): Dataset[T] = withPlan {
+Repartition(numPartitions, shuffle = true, _)
+ }
+
+ /**
+* Returns a new [[Dataset]] that has exactly `numPartitions` partitions.
+* Similar to coalesce defined on an [[RDD]], this operation results in a
narrow dependency, e.g.
+* if you go from 1000 partitions to 100 partitions, there will not be a
shuffle, instead each of
+* the 100 new partitions will claim 10 of the current partitions.
+* @since 1.6.0
+*/
+ def coalesce(numPartitions: Int): Dataset[T] = withPlan {
+Repartition(numPartitions, shuffle = false, _)
+ }
+
/* *** *
* Functional Operations *
* *** */
http://git-wip-us.apache.org/repos/asf/spark/blob/238ae51b/sql/core/src/test/scala/org/apache/spark/sql/DatasetSuite.scala
--
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/DatasetSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/DatasetSuite.scala
index 13eede1..c253fdb 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/DatasetSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/DatasetSuite.scala
@@ -52,6 +52,21 @@ class DatasetSuite extends QueryTest with SharedSQLContext {
assert(ds.takeAsList(1).get(0) == item)
}
+ test("coalesce, repartition") {
+val data = (1 to 100).map(i => ClassData(i.toString, i))
+val ds = data.toDS()
+
+assert(ds.repartition(10).rdd.partitions.length == 10)
+checkAnswer(
+ ds.repartition(10),
+ data: _*)
+
+assert(ds.coalesce(1).rdd.partitions.length == 1)
+checkAnswer(
+ ds.coalesce(1),
+ data: _*)
+ }
+
test("as tuple") {
val data = Seq(("a", 1), ("b", 2)).toDF("a", "b")
checkAnswer(
-
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
spark git commit: Added a line of comment to explain why the extra sort exists in pivot.
Repository: spark Updated Branches: refs/heads/branch-1.6 6d8c4c644 -> 36a99f93f Added a line of comment to explain why the extra sort exists in pivot. (cherry picked from commit 34ca392da7097a1fbe48cd6c3ebff51453ca26ca) Signed-off-by: Reynold Xin Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/36a99f93 Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/36a99f93 Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/36a99f93 Branch: refs/heads/branch-1.6 Commit: 36a99f93f7d77dac1159e5703d931d24cb24b154 Parents: 6d8c4c6 Author: Reynold Xin Authored: Tue Nov 24 14:51:01 2015 -0800 Committer: Reynold Xin Committed: Tue Nov 24 16:23:40 2015 -0800 -- sql/core/src/main/scala/org/apache/spark/sql/GroupedData.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) -- http://git-wip-us.apache.org/repos/asf/spark/blob/36a99f93/sql/core/src/main/scala/org/apache/spark/sql/GroupedData.scala -- diff --git a/sql/core/src/main/scala/org/apache/spark/sql/GroupedData.scala b/sql/core/src/main/scala/org/apache/spark/sql/GroupedData.scala index ee7150c..abd531c 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/GroupedData.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/GroupedData.scala @@ -304,7 +304,7 @@ class GroupedData protected[sql]( // Get the distinct values of the column and sort them so its consistent val values = df.select(pivotColumn) .distinct() - .sort(pivotColumn) + .sort(pivotColumn) // ensure that the output columns are in a consistent logical order .map(_.get(0)) .take(maxValues + 1) .toSeq - To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org
spark git commit: [SPARK-11967][SQL] Consistent use of varargs for multiple paths in DataFrameReader
Repository: spark
Updated Branches:
refs/heads/branch-1.6 36a99f93f -> 4464fa25c
[SPARK-11967][SQL] Consistent use of varargs for multiple paths in
DataFrameReader
This patch makes it consistent to use varargs in all DataFrameReader methods,
including Parquet, JSON, text, and the generic load function.
Also added a few more API tests for the Java API.
Author: Reynold Xin
Closes #9945 from rxin/SPARK-11967.
(cherry picked from commit 25bbd3c16e8e8be4d2c43000223d54650e9a3696)
Signed-off-by: Reynold Xin
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/4464fa25
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/4464fa25
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/4464fa25
Branch: refs/heads/branch-1.6
Commit: 4464fa25c227ce9012862f9940babc47527ebbf3
Parents: 36a99f9
Author: Reynold Xin
Authored: Tue Nov 24 18:16:07 2015 -0800
Committer: Reynold Xin
Committed: Tue Nov 24 18:16:14 2015 -0800
--
python/pyspark/sql/readwriter.py| 19 +++
.../org/apache/spark/sql/DataFrameReader.scala | 36
.../apache/spark/sql/JavaDataFrameSuite.java| 23 +
sql/core/src/test/resources/text-suite2.txt | 1 +
.../org/apache/spark/sql/DataFrameSuite.scala | 2 +-
5 files changed, 66 insertions(+), 15 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/4464fa25/python/pyspark/sql/readwriter.py
--
diff --git a/python/pyspark/sql/readwriter.py b/python/pyspark/sql/readwriter.py
index e8f0d7e..2e75f0c 100644
--- a/python/pyspark/sql/readwriter.py
+++ b/python/pyspark/sql/readwriter.py
@@ -109,7 +109,7 @@ class DataFrameReader(object):
def load(self, path=None, format=None, schema=None, **options):
"""Loads data from a data source and returns it as a :class`DataFrame`.
-:param path: optional string for file-system backed data sources.
+:param path: optional string or a list of string for file-system
backed data sources.
:param format: optional string for format of the data source. Default
to 'parquet'.
:param schema: optional :class:`StructType` for the input schema.
:param options: all other string options
@@ -118,6 +118,7 @@ class DataFrameReader(object):
... opt2=1, opt3='str')
>>> df.dtypes
[('name', 'string'), ('year', 'int'), ('month', 'int'), ('day', 'int')]
+
>>> df =
sqlContext.read.format('json').load(['python/test_support/sql/people.json',
... 'python/test_support/sql/people1.json'])
>>> df.dtypes
@@ -130,10 +131,8 @@ class DataFrameReader(object):
self.options(**options)
if path is not None:
if type(path) == list:
-paths = path
-gateway = self._sqlContext._sc._gateway
-jpaths = utils.toJArray(gateway, gateway.jvm.java.lang.String,
paths)
-return self._df(self._jreader.load(jpaths))
+return self._df(
+
self._jreader.load(self._sqlContext._sc._jvm.PythonUtils.toSeq(path)))
else:
return self._df(self._jreader.load(path))
else:
@@ -175,6 +174,8 @@ class DataFrameReader(object):
self.schema(schema)
if isinstance(path, basestring):
return self._df(self._jreader.json(path))
+elif type(path) == list:
+return
self._df(self._jreader.json(self._sqlContext._sc._jvm.PythonUtils.toSeq(path)))
elif isinstance(path, RDD):
return self._df(self._jreader.json(path._jrdd))
else:
@@ -205,16 +206,20 @@ class DataFrameReader(object):
@ignore_unicode_prefix
@since(1.6)
-def text(self, path):
+def text(self, paths):
"""Loads a text file and returns a [[DataFrame]] with a single string
column named "text".
Each line in the text file is a new row in the resulting DataFrame.
+:param paths: string, or list of strings, for input path(s).
+
>>> df = sqlContext.read.text('python/test_support/sql/text-test.txt')
>>> df.collect()
[Row(value=u'hello'), Row(value=u'this')]
"""
-return self._df(self._jreader.text(path))
+if isinstance(paths, basestring):
+paths = [paths]
+return
self._df(self._jreader.text(self._sqlContext._sc._jvm.PythonUtils.toSeq(paths)))
@since(1.5)
def orc(self, path):
http://git-wip-us.apache.org/repos/asf/spark/blob/4464fa25/sql/core/src/main/scala/org/apache/spark/sql/DataFrameReader.scala
--
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/DataFrameReader.scala
b
spark git commit: [SPARK-11967][SQL] Consistent use of varargs for multiple paths in DataFrameReader
Repository: spark
Updated Branches:
refs/heads/master 238ae51b6 -> 25bbd3c16
[SPARK-11967][SQL] Consistent use of varargs for multiple paths in
DataFrameReader
This patch makes it consistent to use varargs in all DataFrameReader methods,
including Parquet, JSON, text, and the generic load function.
Also added a few more API tests for the Java API.
Author: Reynold Xin
Closes #9945 from rxin/SPARK-11967.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/25bbd3c1
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/25bbd3c1
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/25bbd3c1
Branch: refs/heads/master
Commit: 25bbd3c16e8e8be4d2c43000223d54650e9a3696
Parents: 238ae51
Author: Reynold Xin
Authored: Tue Nov 24 18:16:07 2015 -0800
Committer: Reynold Xin
Committed: Tue Nov 24 18:16:07 2015 -0800
--
python/pyspark/sql/readwriter.py| 19 +++
.../org/apache/spark/sql/DataFrameReader.scala | 36
.../apache/spark/sql/JavaDataFrameSuite.java| 23 +
sql/core/src/test/resources/text-suite2.txt | 1 +
.../org/apache/spark/sql/DataFrameSuite.scala | 2 +-
5 files changed, 66 insertions(+), 15 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/25bbd3c1/python/pyspark/sql/readwriter.py
--
diff --git a/python/pyspark/sql/readwriter.py b/python/pyspark/sql/readwriter.py
index e8f0d7e..2e75f0c 100644
--- a/python/pyspark/sql/readwriter.py
+++ b/python/pyspark/sql/readwriter.py
@@ -109,7 +109,7 @@ class DataFrameReader(object):
def load(self, path=None, format=None, schema=None, **options):
"""Loads data from a data source and returns it as a :class`DataFrame`.
-:param path: optional string for file-system backed data sources.
+:param path: optional string or a list of string for file-system
backed data sources.
:param format: optional string for format of the data source. Default
to 'parquet'.
:param schema: optional :class:`StructType` for the input schema.
:param options: all other string options
@@ -118,6 +118,7 @@ class DataFrameReader(object):
... opt2=1, opt3='str')
>>> df.dtypes
[('name', 'string'), ('year', 'int'), ('month', 'int'), ('day', 'int')]
+
>>> df =
sqlContext.read.format('json').load(['python/test_support/sql/people.json',
... 'python/test_support/sql/people1.json'])
>>> df.dtypes
@@ -130,10 +131,8 @@ class DataFrameReader(object):
self.options(**options)
if path is not None:
if type(path) == list:
-paths = path
-gateway = self._sqlContext._sc._gateway
-jpaths = utils.toJArray(gateway, gateway.jvm.java.lang.String,
paths)
-return self._df(self._jreader.load(jpaths))
+return self._df(
+
self._jreader.load(self._sqlContext._sc._jvm.PythonUtils.toSeq(path)))
else:
return self._df(self._jreader.load(path))
else:
@@ -175,6 +174,8 @@ class DataFrameReader(object):
self.schema(schema)
if isinstance(path, basestring):
return self._df(self._jreader.json(path))
+elif type(path) == list:
+return
self._df(self._jreader.json(self._sqlContext._sc._jvm.PythonUtils.toSeq(path)))
elif isinstance(path, RDD):
return self._df(self._jreader.json(path._jrdd))
else:
@@ -205,16 +206,20 @@ class DataFrameReader(object):
@ignore_unicode_prefix
@since(1.6)
-def text(self, path):
+def text(self, paths):
"""Loads a text file and returns a [[DataFrame]] with a single string
column named "text".
Each line in the text file is a new row in the resulting DataFrame.
+:param paths: string, or list of strings, for input path(s).
+
>>> df = sqlContext.read.text('python/test_support/sql/text-test.txt')
>>> df.collect()
[Row(value=u'hello'), Row(value=u'this')]
"""
-return self._df(self._jreader.text(path))
+if isinstance(paths, basestring):
+paths = [paths]
+return
self._df(self._jreader.text(self._sqlContext._sc._jvm.PythonUtils.toSeq(paths)))
@since(1.5)
def orc(self, path):
http://git-wip-us.apache.org/repos/asf/spark/blob/25bbd3c1/sql/core/src/main/scala/org/apache/spark/sql/DataFrameReader.scala
--
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/DataFrameReader.scala
b/sql/core/src/main/scala/org/apache/spark/sql/DataFrameReader.scala
index dcb3737..3ed1e55 100644
--- a/sq
spark git commit: [SPARK-11947][SQL] Mark deprecated methods with "This will be removed in Spark 2.0."
Repository: spark
Updated Branches:
refs/heads/branch-1.6 4464fa25c -> 862d788fc
[SPARK-11947][SQL] Mark deprecated methods with "This will be removed in Spark
2.0."
Also fixed some documentation as I saw them.
Author: Reynold Xin
Closes #9930 from rxin/SPARK-11947.
(cherry picked from commit 4d6bbbc03ddb6650b00eb638e4876a196014c19c)
Signed-off-by: Reynold Xin
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/862d788f
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/862d788f
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/862d788f
Branch: refs/heads/branch-1.6
Commit: 862d788fce8c4b91e9f821c192b18fc0310a6b41
Parents: 4464fa2
Author: Reynold Xin
Authored: Tue Nov 24 18:58:55 2015 -0800
Committer: Reynold Xin
Committed: Tue Nov 24 18:59:01 2015 -0800
--
project/MimaExcludes.scala | 3 +-
.../scala/org/apache/spark/sql/Column.scala | 20 ++--
.../scala/org/apache/spark/sql/DataFrame.scala | 72 +++
.../scala/org/apache/spark/sql/Dataset.scala| 1 +
.../scala/org/apache/spark/sql/SQLContext.scala | 88 +++---
.../org/apache/spark/sql/SQLImplicits.scala | 25 +++-
.../org/apache/spark/sql/SparkSQLParser.scala | 118 ---
.../spark/sql/execution/SparkSQLParser.scala| 117 ++
.../scala/org/apache/spark/sql/functions.scala | 52
.../SimpleTextHadoopFsRelationSuite.scala | 6 +-
10 files changed, 282 insertions(+), 220 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/862d788f/project/MimaExcludes.scala
--
diff --git a/project/MimaExcludes.scala b/project/MimaExcludes.scala
index bb45d1b..54a9ad9 100644
--- a/project/MimaExcludes.scala
+++ b/project/MimaExcludes.scala
@@ -108,7 +108,8 @@ object MimaExcludes {
ProblemFilters.exclude[MissingClassProblem](
"org.apache.spark.rdd.MapPartitionsWithPreparationRDD"),
ProblemFilters.exclude[MissingClassProblem](
- "org.apache.spark.rdd.MapPartitionsWithPreparationRDD$")
+ "org.apache.spark.rdd.MapPartitionsWithPreparationRDD$"),
+
ProblemFilters.exclude[MissingClassProblem]("org.apache.spark.sql.SparkSQLParser")
) ++ Seq(
// SPARK-11485
ProblemFilters.exclude[MissingMethodProblem]("org.apache.spark.sql.DataFrameHolder.df"),
http://git-wip-us.apache.org/repos/asf/spark/blob/862d788f/sql/core/src/main/scala/org/apache/spark/sql/Column.scala
--
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/Column.scala
b/sql/core/src/main/scala/org/apache/spark/sql/Column.scala
index 30c554a..b3cd9e1 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/Column.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/Column.scala
@@ -42,7 +42,8 @@ private[sql] object Column {
/**
* A [[Column]] where an [[Encoder]] has been given for the expected input and
return type.
- * @since 1.6.0
+ * To create a [[TypedColumn]], use the `as` function on a [[Column]].
+ *
* @tparam T The input type expected for this expression. Can be `Any` if the
expression is type
* checked by the analyzer instead of the compiler (i.e.
`expr("sum(...)")`).
* @tparam U The output type of this column.
@@ -51,7 +52,8 @@ private[sql] object Column {
*/
class TypedColumn[-T, U](
expr: Expression,
-private[sql] val encoder: ExpressionEncoder[U]) extends Column(expr) {
+private[sql] val encoder: ExpressionEncoder[U])
+ extends Column(expr) {
/**
* Inserts the specific input type and schema into any expressions that are
expected to operate
@@ -61,12 +63,11 @@ class TypedColumn[-T, U](
inputEncoder: ExpressionEncoder[_],
schema: Seq[Attribute]): TypedColumn[T, U] = {
val boundEncoder =
inputEncoder.bind(schema).asInstanceOf[ExpressionEncoder[Any]]
-new TypedColumn[T, U] (expr transform {
- case ta: TypedAggregateExpression if ta.aEncoder.isEmpty =>
-ta.copy(
- aEncoder = Some(boundEncoder),
- children = schema)
-}, encoder)
+new TypedColumn[T, U](
+ expr transform { case ta: TypedAggregateExpression if
ta.aEncoder.isEmpty =>
+ta.copy(aEncoder = Some(boundEncoder), children = schema)
+ },
+ encoder)
}
}
@@ -691,8 +692,9 @@ class Column(protected[sql] val expr: Expression) extends
Logging {
*
* @group expr_ops
* @since 1.3.0
+ * @deprecated As of 1.5.0. Use isin. This will be removed in Spark 2.0.
*/
- @deprecated("use isin", "1.5.0")
+ @deprecated("use isin. This will be removed in Spark 2.0.", "1.5.0")
@scala.annotation.varargs
def in(list: Any*): Column = isin(list :
spark git commit: [SPARK-11947][SQL] Mark deprecated methods with "This will be removed in Spark 2.0."
Repository: spark
Updated Branches:
refs/heads/master 25bbd3c16 -> 4d6bbbc03
[SPARK-11947][SQL] Mark deprecated methods with "This will be removed in Spark
2.0."
Also fixed some documentation as I saw them.
Author: Reynold Xin
Closes #9930 from rxin/SPARK-11947.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/4d6bbbc0
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/4d6bbbc0
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/4d6bbbc0
Branch: refs/heads/master
Commit: 4d6bbbc03ddb6650b00eb638e4876a196014c19c
Parents: 25bbd3c
Author: Reynold Xin
Authored: Tue Nov 24 18:58:55 2015 -0800
Committer: Reynold Xin
Committed: Tue Nov 24 18:58:55 2015 -0800
--
project/MimaExcludes.scala | 3 +-
.../scala/org/apache/spark/sql/Column.scala | 20 ++--
.../scala/org/apache/spark/sql/DataFrame.scala | 72 +++
.../scala/org/apache/spark/sql/Dataset.scala| 1 +
.../scala/org/apache/spark/sql/SQLContext.scala | 88 +++---
.../org/apache/spark/sql/SQLImplicits.scala | 25 +++-
.../org/apache/spark/sql/SparkSQLParser.scala | 118 ---
.../spark/sql/execution/SparkSQLParser.scala| 117 ++
.../scala/org/apache/spark/sql/functions.scala | 52
.../SimpleTextHadoopFsRelationSuite.scala | 6 +-
10 files changed, 282 insertions(+), 220 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/4d6bbbc0/project/MimaExcludes.scala
--
diff --git a/project/MimaExcludes.scala b/project/MimaExcludes.scala
index bb45d1b..54a9ad9 100644
--- a/project/MimaExcludes.scala
+++ b/project/MimaExcludes.scala
@@ -108,7 +108,8 @@ object MimaExcludes {
ProblemFilters.exclude[MissingClassProblem](
"org.apache.spark.rdd.MapPartitionsWithPreparationRDD"),
ProblemFilters.exclude[MissingClassProblem](
- "org.apache.spark.rdd.MapPartitionsWithPreparationRDD$")
+ "org.apache.spark.rdd.MapPartitionsWithPreparationRDD$"),
+
ProblemFilters.exclude[MissingClassProblem]("org.apache.spark.sql.SparkSQLParser")
) ++ Seq(
// SPARK-11485
ProblemFilters.exclude[MissingMethodProblem]("org.apache.spark.sql.DataFrameHolder.df"),
http://git-wip-us.apache.org/repos/asf/spark/blob/4d6bbbc0/sql/core/src/main/scala/org/apache/spark/sql/Column.scala
--
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/Column.scala
b/sql/core/src/main/scala/org/apache/spark/sql/Column.scala
index 30c554a..b3cd9e1 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/Column.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/Column.scala
@@ -42,7 +42,8 @@ private[sql] object Column {
/**
* A [[Column]] where an [[Encoder]] has been given for the expected input and
return type.
- * @since 1.6.0
+ * To create a [[TypedColumn]], use the `as` function on a [[Column]].
+ *
* @tparam T The input type expected for this expression. Can be `Any` if the
expression is type
* checked by the analyzer instead of the compiler (i.e.
`expr("sum(...)")`).
* @tparam U The output type of this column.
@@ -51,7 +52,8 @@ private[sql] object Column {
*/
class TypedColumn[-T, U](
expr: Expression,
-private[sql] val encoder: ExpressionEncoder[U]) extends Column(expr) {
+private[sql] val encoder: ExpressionEncoder[U])
+ extends Column(expr) {
/**
* Inserts the specific input type and schema into any expressions that are
expected to operate
@@ -61,12 +63,11 @@ class TypedColumn[-T, U](
inputEncoder: ExpressionEncoder[_],
schema: Seq[Attribute]): TypedColumn[T, U] = {
val boundEncoder =
inputEncoder.bind(schema).asInstanceOf[ExpressionEncoder[Any]]
-new TypedColumn[T, U] (expr transform {
- case ta: TypedAggregateExpression if ta.aEncoder.isEmpty =>
-ta.copy(
- aEncoder = Some(boundEncoder),
- children = schema)
-}, encoder)
+new TypedColumn[T, U](
+ expr transform { case ta: TypedAggregateExpression if
ta.aEncoder.isEmpty =>
+ta.copy(aEncoder = Some(boundEncoder), children = schema)
+ },
+ encoder)
}
}
@@ -691,8 +692,9 @@ class Column(protected[sql] val expr: Expression) extends
Logging {
*
* @group expr_ops
* @since 1.3.0
+ * @deprecated As of 1.5.0. Use isin. This will be removed in Spark 2.0.
*/
- @deprecated("use isin", "1.5.0")
+ @deprecated("use isin. This will be removed in Spark 2.0.", "1.5.0")
@scala.annotation.varargs
def in(list: Any*): Column = isin(list : _*)
http://git-wip-us.apache.org/repos/asf/spark/blob/4d6bbbc0/sql/core/src/main/scala/org/apache/spar
spark git commit: [STREAMING][FLAKY-TEST] Catch execution context race condition in `FileBasedWriteAheadLog.close()`
Repository: spark
Updated Branches:
refs/heads/master 4d6bbbc03 -> a5d988763
[STREAMING][FLAKY-TEST] Catch execution context race condition in
`FileBasedWriteAheadLog.close()`
There is a race condition in `FileBasedWriteAheadLog.close()`, where if
delete's of old log files are in progress, the write ahead log may close, and
result in a `RejectedExecutionException`. This is okay, and should be handled
gracefully.
Example test failures:
https://amplab.cs.berkeley.edu/jenkins/job/Spark-1.6-SBT/AMPLAB_JENKINS_BUILD_PROFILE=hadoop1.0,label=spark-test/95/testReport/junit/org.apache.spark.streaming.util/BatchedWriteAheadLogWithCloseFileAfterWriteSuite/BatchedWriteAheadLog___clean_old_logs/
The reason the test fails is in `afterEach`, `writeAheadLog.close` is called,
and there may still be async deletes in flight.
tdas zsxwing
Author: Burak Yavuz
Closes #9953 from brkyvz/flaky-ss.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/a5d98876
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/a5d98876
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/a5d98876
Branch: refs/heads/master
Commit: a5d988763319f63a8e2b58673dd4f9098f17c835
Parents: 4d6bbbc
Author: Burak Yavuz
Authored: Tue Nov 24 20:58:47 2015 -0800
Committer: Shixiong Zhu
Committed: Tue Nov 24 20:58:47 2015 -0800
--
.../streaming/util/FileBasedWriteAheadLog.scala | 16 +++-
1 file changed, 11 insertions(+), 5 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/a5d98876/streaming/src/main/scala/org/apache/spark/streaming/util/FileBasedWriteAheadLog.scala
--
diff --git
a/streaming/src/main/scala/org/apache/spark/streaming/util/FileBasedWriteAheadLog.scala
b/streaming/src/main/scala/org/apache/spark/streaming/util/FileBasedWriteAheadLog.scala
index 72705f1..f5165f7 100644
---
a/streaming/src/main/scala/org/apache/spark/streaming/util/FileBasedWriteAheadLog.scala
+++
b/streaming/src/main/scala/org/apache/spark/streaming/util/FileBasedWriteAheadLog.scala
@@ -17,7 +17,7 @@
package org.apache.spark.streaming.util
import java.nio.ByteBuffer
-import java.util.concurrent.ThreadPoolExecutor
+import java.util.concurrent.{RejectedExecutionException, ThreadPoolExecutor}
import java.util.{Iterator => JIterator}
import scala.collection.JavaConverters._
@@ -176,10 +176,16 @@ private[streaming] class FileBasedWriteAheadLog(
}
oldLogFiles.foreach { logInfo =>
if (!executionContext.isShutdown) {
-val f = Future { deleteFile(logInfo) }(executionContext)
-if (waitForCompletion) {
- import scala.concurrent.duration._
- Await.ready(f, 1 second)
+try {
+ val f = Future { deleteFile(logInfo) }(executionContext)
+ if (waitForCompletion) {
+import scala.concurrent.duration._
+Await.ready(f, 1 second)
+ }
+} catch {
+ case e: RejectedExecutionException =>
+logWarning("Execution context shutdown before deleting old
WriteAheadLogs. " +
+ "This would not affect recovery correctness.", e)
}
}
}
-
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
spark git commit: [STREAMING][FLAKY-TEST] Catch execution context race condition in `FileBasedWriteAheadLog.close()`
Repository: spark
Updated Branches:
refs/heads/branch-1.6 862d788fc -> b18112666
[STREAMING][FLAKY-TEST] Catch execution context race condition in
`FileBasedWriteAheadLog.close()`
There is a race condition in `FileBasedWriteAheadLog.close()`, where if
delete's of old log files are in progress, the write ahead log may close, and
result in a `RejectedExecutionException`. This is okay, and should be handled
gracefully.
Example test failures:
https://amplab.cs.berkeley.edu/jenkins/job/Spark-1.6-SBT/AMPLAB_JENKINS_BUILD_PROFILE=hadoop1.0,label=spark-test/95/testReport/junit/org.apache.spark.streaming.util/BatchedWriteAheadLogWithCloseFileAfterWriteSuite/BatchedWriteAheadLog___clean_old_logs/
The reason the test fails is in `afterEach`, `writeAheadLog.close` is called,
and there may still be async deletes in flight.
tdas zsxwing
Author: Burak Yavuz
Closes #9953 from brkyvz/flaky-ss.
(cherry picked from commit a5d988763319f63a8e2b58673dd4f9098f17c835)
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/b1811266
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/b1811266
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/b1811266
Branch: refs/heads/branch-1.6
Commit: b18112666adec0a942d1cfe8d6b9f1e7c7201fcd
Parents: 862d788
Author: Burak Yavuz
Authored: Tue Nov 24 20:58:47 2015 -0800
Committer: Shixiong Zhu
Committed: Tue Nov 24 20:59:01 2015 -0800
--
.../streaming/util/FileBasedWriteAheadLog.scala | 16 +++-
1 file changed, 11 insertions(+), 5 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/b1811266/streaming/src/main/scala/org/apache/spark/streaming/util/FileBasedWriteAheadLog.scala
--
diff --git
a/streaming/src/main/scala/org/apache/spark/streaming/util/FileBasedWriteAheadLog.scala
b/streaming/src/main/scala/org/apache/spark/streaming/util/FileBasedWriteAheadLog.scala
index 72705f1..f5165f7 100644
---
a/streaming/src/main/scala/org/apache/spark/streaming/util/FileBasedWriteAheadLog.scala
+++
b/streaming/src/main/scala/org/apache/spark/streaming/util/FileBasedWriteAheadLog.scala
@@ -17,7 +17,7 @@
package org.apache.spark.streaming.util
import java.nio.ByteBuffer
-import java.util.concurrent.ThreadPoolExecutor
+import java.util.concurrent.{RejectedExecutionException, ThreadPoolExecutor}
import java.util.{Iterator => JIterator}
import scala.collection.JavaConverters._
@@ -176,10 +176,16 @@ private[streaming] class FileBasedWriteAheadLog(
}
oldLogFiles.foreach { logInfo =>
if (!executionContext.isShutdown) {
-val f = Future { deleteFile(logInfo) }(executionContext)
-if (waitForCompletion) {
- import scala.concurrent.duration._
- Await.ready(f, 1 second)
+try {
+ val f = Future { deleteFile(logInfo) }(executionContext)
+ if (waitForCompletion) {
+import scala.concurrent.duration._
+Await.ready(f, 1 second)
+ }
+} catch {
+ case e: RejectedExecutionException =>
+logWarning("Execution context shutdown before deleting old
WriteAheadLogs. " +
+ "This would not affect recovery correctness.", e)
}
}
}
-
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
spark git commit: [SPARK-10621][SQL] Consistent naming for functions in SQL, Python, Scala
Repository: spark
Updated Branches:
refs/heads/branch-1.6 b18112666 -> 486db8789
[SPARK-10621][SQL] Consistent naming for functions in SQL, Python, Scala
Author: Reynold Xin
Closes #9948 from rxin/SPARK-10621.
(cherry picked from commit 151d7c2baf18403e6e59e97c80c8bcded6148038)
Signed-off-by: Reynold Xin
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/486db878
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/486db878
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/486db878
Branch: refs/heads/branch-1.6
Commit: 486db87899bcd3d0ff222c8a31d572fe836b69e7
Parents: b181126
Author: Reynold Xin
Authored: Tue Nov 24 21:30:53 2015 -0800
Committer: Reynold Xin
Committed: Tue Nov 24 21:30:59 2015 -0800
--
python/pyspark/sql/functions.py | 111 ++---
.../scala/org/apache/spark/sql/functions.scala | 124 +++
2 files changed, 196 insertions(+), 39 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/486db878/python/pyspark/sql/functions.py
--
diff --git a/python/pyspark/sql/functions.py b/python/pyspark/sql/functions.py
index a1ca723..e3786e0 100644
--- a/python/pyspark/sql/functions.py
+++ b/python/pyspark/sql/functions.py
@@ -150,18 +150,18 @@ _binary_mathfunctions = {
_window_functions = {
'rowNumber':
-"""returns a sequential number starting at 1 within a window partition.
-
-This is equivalent to the ROW_NUMBER function in SQL.""",
+""".. note:: Deprecated in 1.6, use row_number instead.""",
+'row_number':
+"""returns a sequential number starting at 1 within a window
partition.""",
'denseRank':
+""".. note:: Deprecated in 1.6, use dense_rank instead.""",
+'dense_rank':
"""returns the rank of rows within a window partition, without any
gaps.
The difference between rank and denseRank is that denseRank leaves no
gaps in ranking
sequence when there are ties. That is, if you were ranking a
competition using denseRank
and had three people tie for second place, you would say that all
three were in second
-place and that the next person came in third.
-
-This is equivalent to the DENSE_RANK function in SQL.""",
+place and that the next person came in third.""",
'rank':
"""returns the rank of rows within a window partition.
@@ -172,14 +172,14 @@ _window_functions = {
This is equivalent to the RANK function in SQL.""",
'cumeDist':
+""".. note:: Deprecated in 1.6, use cume_dist instead.""",
+'cume_dist':
"""returns the cumulative distribution of values within a window
partition,
-i.e. the fraction of rows that are below the current row.
-
-This is equivalent to the CUME_DIST function in SQL.""",
+i.e. the fraction of rows that are below the current row.""",
'percentRank':
-"""returns the relative rank (i.e. percentile) of rows within a window
partition.
-
-This is equivalent to the PERCENT_RANK function in SQL.""",
+""".. note:: Deprecated in 1.6, use percent_rank instead.""",
+'percent_rank':
+"""returns the relative rank (i.e. percentile) of rows within a window
partition.""",
}
for _name, _doc in _functions.items():
@@ -189,7 +189,7 @@ for _name, _doc in _functions_1_4.items():
for _name, _doc in _binary_mathfunctions.items():
globals()[_name] = since(1.4)(_create_binary_mathfunction(_name, _doc))
for _name, _doc in _window_functions.items():
-globals()[_name] = since(1.4)(_create_window_function(_name, _doc))
+globals()[_name] = since(1.6)(_create_window_function(_name, _doc))
for _name, _doc in _functions_1_6.items():
globals()[_name] = since(1.6)(_create_function(_name, _doc))
del _name, _doc
@@ -288,6 +288,38 @@ def countDistinct(col, *cols):
@since(1.4)
def monotonicallyIncreasingId():
+"""
+.. note:: Deprecated in 1.6, use monotonically_increasing_id instead.
+"""
+return monotonically_increasing_id()
+
+
+@since(1.6)
+def input_file_name():
+"""Creates a string column for the file name of the current Spark task.
+"""
+sc = SparkContext._active_spark_context
+return Column(sc._jvm.functions.input_file_name())
+
+
+@since(1.6)
+def isnan(col):
+"""An expression that returns true iff the column is NaN.
+"""
+sc = SparkContext._active_spark_context
+return Column(sc._jvm.functions.isnan(_to_java_column(col)))
+
+
+@since(1.6)
+def isnull(col):
+"""An expression that returns true iff the column is null.
+"""
+sc = SparkContext._active_spark_context
+return Column(sc._jvm.functions.isnull(_to_java_column(col)))
+
+
+@since(1.6)
spark git commit: [SPARK-10621][SQL] Consistent naming for functions in SQL, Python, Scala
Repository: spark
Updated Branches:
refs/heads/master a5d988763 -> 151d7c2ba
[SPARK-10621][SQL] Consistent naming for functions in SQL, Python, Scala
Author: Reynold Xin
Closes #9948 from rxin/SPARK-10621.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/151d7c2b
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/151d7c2b
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/151d7c2b
Branch: refs/heads/master
Commit: 151d7c2baf18403e6e59e97c80c8bcded6148038
Parents: a5d9887
Author: Reynold Xin
Authored: Tue Nov 24 21:30:53 2015 -0800
Committer: Reynold Xin
Committed: Tue Nov 24 21:30:53 2015 -0800
--
python/pyspark/sql/functions.py | 111 ++---
.../scala/org/apache/spark/sql/functions.scala | 124 +++
2 files changed, 196 insertions(+), 39 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/151d7c2b/python/pyspark/sql/functions.py
--
diff --git a/python/pyspark/sql/functions.py b/python/pyspark/sql/functions.py
index a1ca723..e3786e0 100644
--- a/python/pyspark/sql/functions.py
+++ b/python/pyspark/sql/functions.py
@@ -150,18 +150,18 @@ _binary_mathfunctions = {
_window_functions = {
'rowNumber':
-"""returns a sequential number starting at 1 within a window partition.
-
-This is equivalent to the ROW_NUMBER function in SQL.""",
+""".. note:: Deprecated in 1.6, use row_number instead.""",
+'row_number':
+"""returns a sequential number starting at 1 within a window
partition.""",
'denseRank':
+""".. note:: Deprecated in 1.6, use dense_rank instead.""",
+'dense_rank':
"""returns the rank of rows within a window partition, without any
gaps.
The difference between rank and denseRank is that denseRank leaves no
gaps in ranking
sequence when there are ties. That is, if you were ranking a
competition using denseRank
and had three people tie for second place, you would say that all
three were in second
-place and that the next person came in third.
-
-This is equivalent to the DENSE_RANK function in SQL.""",
+place and that the next person came in third.""",
'rank':
"""returns the rank of rows within a window partition.
@@ -172,14 +172,14 @@ _window_functions = {
This is equivalent to the RANK function in SQL.""",
'cumeDist':
+""".. note:: Deprecated in 1.6, use cume_dist instead.""",
+'cume_dist':
"""returns the cumulative distribution of values within a window
partition,
-i.e. the fraction of rows that are below the current row.
-
-This is equivalent to the CUME_DIST function in SQL.""",
+i.e. the fraction of rows that are below the current row.""",
'percentRank':
-"""returns the relative rank (i.e. percentile) of rows within a window
partition.
-
-This is equivalent to the PERCENT_RANK function in SQL.""",
+""".. note:: Deprecated in 1.6, use percent_rank instead.""",
+'percent_rank':
+"""returns the relative rank (i.e. percentile) of rows within a window
partition.""",
}
for _name, _doc in _functions.items():
@@ -189,7 +189,7 @@ for _name, _doc in _functions_1_4.items():
for _name, _doc in _binary_mathfunctions.items():
globals()[_name] = since(1.4)(_create_binary_mathfunction(_name, _doc))
for _name, _doc in _window_functions.items():
-globals()[_name] = since(1.4)(_create_window_function(_name, _doc))
+globals()[_name] = since(1.6)(_create_window_function(_name, _doc))
for _name, _doc in _functions_1_6.items():
globals()[_name] = since(1.6)(_create_function(_name, _doc))
del _name, _doc
@@ -288,6 +288,38 @@ def countDistinct(col, *cols):
@since(1.4)
def monotonicallyIncreasingId():
+"""
+.. note:: Deprecated in 1.6, use monotonically_increasing_id instead.
+"""
+return monotonically_increasing_id()
+
+
+@since(1.6)
+def input_file_name():
+"""Creates a string column for the file name of the current Spark task.
+"""
+sc = SparkContext._active_spark_context
+return Column(sc._jvm.functions.input_file_name())
+
+
+@since(1.6)
+def isnan(col):
+"""An expression that returns true iff the column is NaN.
+"""
+sc = SparkContext._active_spark_context
+return Column(sc._jvm.functions.isnan(_to_java_column(col)))
+
+
+@since(1.6)
+def isnull(col):
+"""An expression that returns true iff the column is null.
+"""
+sc = SparkContext._active_spark_context
+return Column(sc._jvm.functions.isnull(_to_java_column(col)))
+
+
+@since(1.6)
+def monotonically_increasing_id():
"""A column that generates monotonically increasing 64-bit integ
spark git commit: [SPARK-11140][CORE] Transfer files using network lib when using NettyRpcEnv - 1.6.version.
Repository: spark
Updated Branches:
refs/heads/branch-1.6 486db8789 -> 68bcb9b33
[SPARK-11140][CORE] Transfer files using network lib when using NettyRpcEnv -
1.6.version.
This patch is the same code as in SPARK-11140 in master, but with some added
code to still use the HTTP file server by default in NettyRpcEnv. This is
mostly to avoid conflicts when backporting patches to 1.6.
Author: Marcelo Vanzin
Closes #9947 from vanzin/SPARK-11140-branch-1.6.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/68bcb9b3
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/68bcb9b3
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/68bcb9b3
Branch: refs/heads/branch-1.6
Commit: 68bcb9b33b731af33f6c9444c8c2fc54e50b3202
Parents: 486db87
Author: Marcelo Vanzin
Authored: Tue Nov 24 21:48:51 2015 -0800
Committer: Reynold Xin
Committed: Tue Nov 24 21:48:51 2015 -0800
--
.../scala/org/apache/spark/SparkContext.scala | 8 +-
.../main/scala/org/apache/spark/SparkEnv.scala | 14 --
.../scala/org/apache/spark/rpc/RpcEnv.scala | 46 ++
.../org/apache/spark/rpc/akka/AkkaRpcEnv.scala | 60 +++-
.../spark/rpc/netty/HttpBasedFileServer.scala | 59
.../apache/spark/rpc/netty/NettyRpcEnv.scala| 147 +--
.../spark/rpc/netty/NettyStreamManager.scala| 63
.../scala/org/apache/spark/util/Utils.scala | 9 ++
.../org/apache/spark/rpc/RpcEnvSuite.scala | 39 -
.../spark/rpc/netty/NettyRpcHandlerSuite.scala | 10 +-
.../spark/launcher/AbstractCommandBuilder.java | 2 +-
.../network/client/TransportClientFactory.java | 6 +-
.../network/server/TransportChannelHandler.java | 1 +
13 files changed, 418 insertions(+), 46 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/68bcb9b3/core/src/main/scala/org/apache/spark/SparkContext.scala
--
diff --git a/core/src/main/scala/org/apache/spark/SparkContext.scala
b/core/src/main/scala/org/apache/spark/SparkContext.scala
index 90480e5..e19ba11 100644
--- a/core/src/main/scala/org/apache/spark/SparkContext.scala
+++ b/core/src/main/scala/org/apache/spark/SparkContext.scala
@@ -1379,7 +1379,7 @@ class SparkContext(config: SparkConf) extends Logging
with ExecutorAllocationCli
}
val key = if (!isLocal && scheme == "file") {
- env.httpFileServer.addFile(new File(uri.getPath))
+ env.rpcEnv.fileServer.addFile(new File(uri.getPath))
} else {
schemeCorrectedPath
}
@@ -1630,7 +1630,7 @@ class SparkContext(config: SparkConf) extends Logging
with ExecutorAllocationCli
var key = ""
if (path.contains("\\")) {
// For local paths with backslashes on Windows, URI throws an exception
-key = env.httpFileServer.addJar(new File(path))
+key = env.rpcEnv.fileServer.addJar(new File(path))
} else {
val uri = new URI(path)
key = uri.getScheme match {
@@ -1644,7 +1644,7 @@ class SparkContext(config: SparkConf) extends Logging
with ExecutorAllocationCli
// of the AM to make it show up in the current working directory.
val fileName = new Path(uri.getPath).getName()
try {
-env.httpFileServer.addJar(new File(fileName))
+env.rpcEnv.fileServer.addJar(new File(fileName))
} catch {
case e: Exception =>
// For now just log an error but allow to go through so
spark examples work.
@@ -1655,7 +1655,7 @@ class SparkContext(config: SparkConf) extends Logging
with ExecutorAllocationCli
}
} else {
try {
-env.httpFileServer.addJar(new File(uri.getPath))
+env.rpcEnv.fileServer.addJar(new File(uri.getPath))
} catch {
case exc: FileNotFoundException =>
logError(s"Jar not found at $path")
http://git-wip-us.apache.org/repos/asf/spark/blob/68bcb9b3/core/src/main/scala/org/apache/spark/SparkEnv.scala
--
diff --git a/core/src/main/scala/org/apache/spark/SparkEnv.scala
b/core/src/main/scala/org/apache/spark/SparkEnv.scala
index 88df27f..84230e3 100644
--- a/core/src/main/scala/org/apache/spark/SparkEnv.scala
+++ b/core/src/main/scala/org/apache/spark/SparkEnv.scala
@@ -66,7 +66,6 @@ class SparkEnv (
val blockTransferService: BlockTransferService,
val blockManager: BlockManager,
val securityManager: SecurityManager,
-val httpFileServer: HttpFileServer,
val sparkFilesDir: String,
val metricsSystem: MetricsSystem,
val memoryManager: MemoryManager,
@@ -91,7 +90,6 @@ class SparkEnv (
if (!isStopped) {
spark git commit: [SPARK-11979][STREAMING] Empty TrackStateRDD cannot be checkpointed and recovered from checkpoint file
Repository: spark Updated Branches: refs/heads/master 151d7c2ba -> 216988688 [SPARK-11979][STREAMING] Empty TrackStateRDD cannot be checkpointed and recovered from checkpoint file This solves the following exception caused when empty state RDD is checkpointed and recovered. The root cause is that an empty OpenHashMapBasedStateMap cannot be deserialized as the initialCapacity is set to zero. ``` Job aborted due to stage failure: Task 0 in stage 6.0 failed 1 times, most recent failure: Lost task 0.0 in stage 6.0 (TID 20, localhost): java.lang.IllegalArgumentException: requirement failed: Invalid initial capacity at scala.Predef$.require(Predef.scala:233) at org.apache.spark.streaming.util.OpenHashMapBasedStateMap.(StateMap.scala:96) at org.apache.spark.streaming.util.OpenHashMapBasedStateMap.(StateMap.scala:86) at org.apache.spark.streaming.util.OpenHashMapBasedStateMap.readObject(StateMap.scala:291) at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57) at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) at java.lang.reflect.Method.invoke(Method.java:606) at java.io.ObjectStreamClass.invokeReadObject(ObjectStreamClass.java:1017) at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1893) at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798) at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350) at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1990) at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1915) at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798) at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350) at java.io.ObjectInputStream.readObject(ObjectInputStream.java:370) at org.apache.spark.serializer.JavaDeserializationStream.readObject(JavaSerializer.scala:76) at org.apache.spark.serializer.DeserializationStream$$anon$1.getNext(Serializer.scala:181) at org.apache.spark.util.NextIterator.hasNext(NextIterator.scala:73) at scala.collection.Iterator$$anon$13.hasNext(Iterator.scala:371) at scala.collection.Iterator$class.foreach(Iterator.scala:727) at scala.collection.AbstractIterator.foreach(Iterator.scala:1157) at scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:48) at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:103) at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:47) at scala.collection.TraversableOnce$class.to(TraversableOnce.scala:273) at scala.collection.AbstractIterator.to(Iterator.scala:1157) at scala.collection.TraversableOnce$class.toBuffer(TraversableOnce.scala:265) at scala.collection.AbstractIterator.toBuffer(Iterator.scala:1157) at scala.collection.TraversableOnce$class.toArray(TraversableOnce.scala:252) at scala.collection.AbstractIterator.toArray(Iterator.scala:1157) at org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$12.apply(RDD.scala:921) at org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$12.apply(RDD.scala:921) at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1858) at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1858) at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66) at org.apache.spark.scheduler.Task.run(Task.scala:88) at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) at java.lang.Thread.run(Thread.java:744) ``` Author: Tathagata Das Closes #9958 from tdas/SPARK-11979. Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/21698868 Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/21698868 Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/21698868 Branch: refs/heads/master Commit: 2169886883d33b33acf378ac42a626576b342df1 Parents: 151d7c2 Author: Tathagata Das Authored: Tue Nov 24 23:13:01 2015 -0800 Committer: Shixiong Zhu Committed: Tue Nov 24 23:13:01 2015 -0800 -- .../apache/spark/streaming/util/StateMap.scala | 19 - .../apache/spark/streaming/StateMapSuite.scala | 30 +--- .../streaming/rdd/TrackStateRDDSuite.scala | 10 +++ 3 files changed, 42 insertions(+), 17 deletions(-) --
spark git commit: [SPARK-11979][STREAMING] Empty TrackStateRDD cannot be checkpointed and recovered from checkpoint file
Repository: spark Updated Branches: refs/heads/branch-1.6 68bcb9b33 -> 7f030aa42 [SPARK-11979][STREAMING] Empty TrackStateRDD cannot be checkpointed and recovered from checkpoint file This solves the following exception caused when empty state RDD is checkpointed and recovered. The root cause is that an empty OpenHashMapBasedStateMap cannot be deserialized as the initialCapacity is set to zero. ``` Job aborted due to stage failure: Task 0 in stage 6.0 failed 1 times, most recent failure: Lost task 0.0 in stage 6.0 (TID 20, localhost): java.lang.IllegalArgumentException: requirement failed: Invalid initial capacity at scala.Predef$.require(Predef.scala:233) at org.apache.spark.streaming.util.OpenHashMapBasedStateMap.(StateMap.scala:96) at org.apache.spark.streaming.util.OpenHashMapBasedStateMap.(StateMap.scala:86) at org.apache.spark.streaming.util.OpenHashMapBasedStateMap.readObject(StateMap.scala:291) at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57) at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) at java.lang.reflect.Method.invoke(Method.java:606) at java.io.ObjectStreamClass.invokeReadObject(ObjectStreamClass.java:1017) at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1893) at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798) at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350) at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1990) at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1915) at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798) at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350) at java.io.ObjectInputStream.readObject(ObjectInputStream.java:370) at org.apache.spark.serializer.JavaDeserializationStream.readObject(JavaSerializer.scala:76) at org.apache.spark.serializer.DeserializationStream$$anon$1.getNext(Serializer.scala:181) at org.apache.spark.util.NextIterator.hasNext(NextIterator.scala:73) at scala.collection.Iterator$$anon$13.hasNext(Iterator.scala:371) at scala.collection.Iterator$class.foreach(Iterator.scala:727) at scala.collection.AbstractIterator.foreach(Iterator.scala:1157) at scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:48) at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:103) at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:47) at scala.collection.TraversableOnce$class.to(TraversableOnce.scala:273) at scala.collection.AbstractIterator.to(Iterator.scala:1157) at scala.collection.TraversableOnce$class.toBuffer(TraversableOnce.scala:265) at scala.collection.AbstractIterator.toBuffer(Iterator.scala:1157) at scala.collection.TraversableOnce$class.toArray(TraversableOnce.scala:252) at scala.collection.AbstractIterator.toArray(Iterator.scala:1157) at org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$12.apply(RDD.scala:921) at org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$12.apply(RDD.scala:921) at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1858) at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1858) at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66) at org.apache.spark.scheduler.Task.run(Task.scala:88) at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) at java.lang.Thread.run(Thread.java:744) ``` Author: Tathagata Das Closes #9958 from tdas/SPARK-11979. (cherry picked from commit 2169886883d33b33acf378ac42a626576b342df1) 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/7f030aa4 Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/7f030aa4 Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/7f030aa4 Branch: refs/heads/branch-1.6 Commit: 7f030aa422802a8e7077e1c74a59ab9a5fe54488 Parents: 68bcb9b Author: Tathagata Das Authored: Tue Nov 24 23:13:01 2015 -0800 Committer: Shixiong Zhu Committed: Tue Nov 24 23:13:29 2015 -0800 -- .../apache/spark/streaming/util/StateMap.scala | 19 - .../apache/spark/streaming/StateMapSuite.scala | 30 +--- .../streaming/rdd/TrackStateRDDSuite.scala | 10 +++ 3 files changed
