You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by rdblue <gi...@git.apache.org> on 2017/09/29 23:54:45 UTC
[GitHub] spark pull request #19394: SPARK-22170: Reduce memory consumption in broadca...
GitHub user rdblue opened a pull request:
https://github.com/apache/spark/pull/19394
SPARK-22170: Reduce memory consumption in broadcast joins.
This updates the broadcast join code path to lazily decompress pages and
iterate through UnsafeRows to prevent all rows from being held in memory
while the broadcast table is being built.
## What changes were proposed in this pull request?
(Please fill in changes proposed in this fix)
## How was this patch tested?
(Please explain how this patch was tested. E.g. unit tests, integration tests, manual tests)
(If this patch involves UI changes, please attach a screenshot; otherwise, remove this)
Please review http://spark.apache.org/contributing.html before opening a pull request.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/rdblue/spark broadcast-driver-memory
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/19394.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #19394
----
commit d82ffca19c84589dfcb87a81996fb3ce2e31e91a
Author: Ryan Blue <bl...@apache.org>
Date: 2017-08-11T00:41:56Z
SPARK-22170: Reduce memory consumption in broadcast joins.
This updates the broadcast join code path to lazily decompress pages and
iterate through UnsafeRows to prevent all rows from being held in memory
while the broadcast table is being built.
----
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/19394
Merged build finished. Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #19394: [SPARK-22170][SQL] Reduce memory consumption in b...
Posted by rdblue <gi...@git.apache.org>.
Github user rdblue commented on a diff in the pull request:
https://github.com/apache/spark/pull/19394#discussion_r143317686
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/SparkPlan.scala ---
@@ -280,13 +280,20 @@ abstract class SparkPlan extends QueryPlan[SparkPlan] with Logging with Serializ
results.toArray
}
+ private[spark] def executeCollectIterator(): (Long, Iterator[InternalRow]) = {
+ val countsAndBytes = getByteArrayRdd().collect()
+ val total = countsAndBytes.map(_._1).sum
+ val rows = countsAndBytes.iterator.map(_._2).flatMap(decodeUnsafeRows)
+ (total, rows)
+ }
+
/**
* Runs this query returning the result as an iterator of InternalRow.
*
* @note Triggers multiple jobs (one for each partition).
*/
def executeToIterator(): Iterator[InternalRow] = {
- getByteArrayRdd().toLocalIterator.flatMap(decodeUnsafeRows)
+ getByteArrayRdd().toLocalIterator.map(_._2).flatMap(decodeUnsafeRows)
--- End diff --
Fixed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #19394: [SPARK-22170][SQL] Reduce memory consumption in b...
Posted by rdblue <gi...@git.apache.org>.
Github user rdblue commented on a diff in the pull request:
https://github.com/apache/spark/pull/19394#discussion_r143517522
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/SparkPlan.scala ---
@@ -274,19 +274,26 @@ abstract class SparkPlan extends QueryPlan[SparkPlan] with Logging with Serializ
val byteArrayRdd = getByteArrayRdd()
val results = ArrayBuffer[InternalRow]()
- byteArrayRdd.collect().foreach { bytes =>
- decodeUnsafeRows(bytes).foreach(results.+=)
+ byteArrayRdd.collect().foreach { rdd =>
+ decodeUnsafeRows(rdd._2).foreach(results.+=)
}
results.toArray
}
+ private[spark] def executeCollectIterator(): (Long, Iterator[InternalRow]) = {
+ val countsAndBytes = getByteArrayRdd().collect()
+ val total = countsAndBytes.map(_._1).sum
+ val rows = countsAndBytes.iterator.flatMap(rdd => decodeUnsafeRows(rdd._2))
+ (total, rows)
+ }
+
/**
* Runs this query returning the result as an iterator of InternalRow.
*
* @note Triggers multiple jobs (one for each partition).
*/
def executeToIterator(): Iterator[InternalRow] = {
- getByteArrayRdd().toLocalIterator.flatMap(decodeUnsafeRows)
+ getByteArrayRdd().toLocalIterator.flatMap(rdd => decodeUnsafeRows(rdd._2))
--- End diff --
Fixed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #19394: [SPARK-22170][SQL] Reduce memory consumption in b...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/19394#discussion_r143094078
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/BroadcastExchangeExec.scala ---
@@ -73,25 +73,37 @@ case class BroadcastExchangeExec(
try {
val beforeCollect = System.nanoTime()
// Note that we use .executeCollect() because we don't want to convert data to Scala types
- val input: Array[InternalRow] = child.executeCollect()
- if (input.length >= 512000000) {
+ val (numRows, input) = child.executeCollectIterator()
+ if (numRows >= 512000000) {
throw new SparkException(
- s"Cannot broadcast the table with more than 512 millions rows: ${input.length} rows")
+ s"Cannot broadcast the table with more than 512 millions rows: $numRows rows")
}
+
val beforeBuild = System.nanoTime()
longMetric("collectTime") += (beforeBuild - beforeCollect) / 1000000
- val dataSize = input.map(_.asInstanceOf[UnsafeRow].getSizeInBytes.toLong).sum
+
+ // Construct the relation.
+ val relation = mode.transform(input, Some(numRows))
+
+ val dataSize = relation match {
+ case map: HashedRelation =>
+ map.estimatedSize
+ case arr: Array[InternalRow] =>
--- End diff --
`mode` is a `BroadcastMode`. So I think we won't hit this case?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/19394
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #19394: [SPARK-22170][SQL] Reduce memory consumption in b...
Posted by rdblue <gi...@git.apache.org>.
Github user rdblue commented on a diff in the pull request:
https://github.com/apache/spark/pull/19394#discussion_r143517490
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/SparkPlan.scala ---
@@ -274,19 +274,26 @@ abstract class SparkPlan extends QueryPlan[SparkPlan] with Logging with Serializ
val byteArrayRdd = getByteArrayRdd()
val results = ArrayBuffer[InternalRow]()
- byteArrayRdd.collect().foreach { bytes =>
- decodeUnsafeRows(bytes).foreach(results.+=)
+ byteArrayRdd.collect().foreach { rdd =>
+ decodeUnsafeRows(rdd._2).foreach(results.+=)
}
results.toArray
}
+ private[spark] def executeCollectIterator(): (Long, Iterator[InternalRow]) = {
+ val countsAndBytes = getByteArrayRdd().collect()
+ val total = countsAndBytes.map(_._1).sum
+ val rows = countsAndBytes.iterator.flatMap(rdd => decodeUnsafeRows(rdd._2))
--- End diff --
Fixed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #19394: [SPARK-22170][SQL] Reduce memory consumption in b...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on a diff in the pull request:
https://github.com/apache/spark/pull/19394#discussion_r143315153
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/SparkPlan.scala ---
@@ -280,13 +280,20 @@ abstract class SparkPlan extends QueryPlan[SparkPlan] with Logging with Serializ
results.toArray
}
+ private[spark] def executeCollectIterator(): (Long, Iterator[InternalRow]) = {
+ val countsAndBytes = getByteArrayRdd().collect()
+ val total = countsAndBytes.map(_._1).sum
+ val rows = countsAndBytes.iterator.map(_._2).flatMap(decodeUnsafeRows)
+ (total, rows)
+ }
+
/**
* Runs this query returning the result as an iterator of InternalRow.
*
* @note Triggers multiple jobs (one for each partition).
*/
def executeToIterator(): Iterator[InternalRow] = {
- getByteArrayRdd().toLocalIterator.flatMap(decodeUnsafeRows)
+ getByteArrayRdd().toLocalIterator.map(_._2).flatMap(decodeUnsafeRows)
--- End diff --
The same here.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on the issue:
https://github.com/apache/spark/pull/19394
Not sure - maybe print the chi-value of the test and see if they make sense. If they do, we can change the threshold.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/19394
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/82524/
Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #19394: [SPARK-22170][SQL] Reduce memory consumption in b...
Posted by rdblue <gi...@git.apache.org>.
Github user rdblue commented on a diff in the pull request:
https://github.com/apache/spark/pull/19394#discussion_r143317518
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/SparkPlan.scala ---
@@ -280,13 +280,20 @@ abstract class SparkPlan extends QueryPlan[SparkPlan] with Logging with Serializ
results.toArray
}
+ private[spark] def executeCollectIterator(): (Long, Iterator[InternalRow]) = {
+ val countsAndBytes = getByteArrayRdd().collect()
+ val total = countsAndBytes.map(_._1).sum
+ val rows = countsAndBytes.iterator.map(_._2).flatMap(decodeUnsafeRows)
+ (total, rows)
+ }
+
/**
* Runs this query returning the result as an iterator of InternalRow.
*
* @note Triggers multiple jobs (one for each partition).
*/
def executeToIterator(): Iterator[InternalRow] = {
- getByteArrayRdd().toLocalIterator.flatMap(decodeUnsafeRows)
+ getByteArrayRdd().toLocalIterator.map(_._2).flatMap(decodeUnsafeRows)
--- End diff --
Good point. I guess from the tests that this will create a new RDD, which isn't what we want.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by rdblue <gi...@git.apache.org>.
Github user rdblue commented on the issue:
https://github.com/apache/spark/pull/19394
Thanks for reviewing, @gatorsmile!
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #19394: [SPARK-22170][SQL] Reduce memory consumption in b...
Posted by rdblue <gi...@git.apache.org>.
Github user rdblue commented on a diff in the pull request:
https://github.com/apache/spark/pull/19394#discussion_r142474766
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/SparkPlan.scala ---
@@ -280,13 +280,20 @@ abstract class SparkPlan extends QueryPlan[SparkPlan] with Logging with Serializ
results.toArray
}
+ private[spark] def executeCollectIterator(): (Long, Iterator[InternalRow]) = {
+ val countsAndBytes = getByteArrayRdd().collect()
--- End diff --
I agree, but this is better than before. This only keeps all of the rows in memory compressed, and then streams through the compressed blocks. Before this patch, the rows are copied into a buffer per row while holding the compressed blocks, so it held the rows compressed and uncompressed at the same time. The uncompressed rows are what this fixes, we can follow up with something better to stream through blocks from executors.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/19394
**[Test build #82556 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/82556/testReport)** for PR 19394 at commit [`1a813ac`](https://github.com/apache/spark/commit/1a813acecb61933544c7c6ca1ebd58d5b95b7fa5).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #19394: [SPARK-22170][SQL] Reduce memory consumption in b...
Posted by kiszk <gi...@git.apache.org>.
Github user kiszk commented on a diff in the pull request:
https://github.com/apache/spark/pull/19394#discussion_r143205255
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/BroadcastExchangeExec.scala ---
@@ -73,25 +73,37 @@ case class BroadcastExchangeExec(
try {
val beforeCollect = System.nanoTime()
// Note that we use .executeCollect() because we don't want to convert data to Scala types
- val input: Array[InternalRow] = child.executeCollect()
- if (input.length >= 512000000) {
+ val (numRows, input) = child.executeCollectIterator()
+ if (numRows >= 512000000) {
throw new SparkException(
- s"Cannot broadcast the table with more than 512 millions rows: ${input.length} rows")
+ s"Cannot broadcast the table with more than 512 millions rows: $numRows rows")
}
+
val beforeBuild = System.nanoTime()
longMetric("collectTime") += (beforeBuild - beforeCollect) / 1000000
- val dataSize = input.map(_.asInstanceOf[UnsafeRow].getSizeInBytes.toLong).sum
+
+ // Construct the relation.
+ val relation = mode.transform(input, Some(numRows))
+
+ val dataSize = relation match {
+ case map: HashedRelation =>
+ map.estimatedSize
+ case arr: Array[InternalRow] =>
--- End diff --
[`IdentityBroadcastMode.transform`](https://github.com/apache/spark/blob/master/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/broadcastMode.scala#L37) seems to return `Array[InternalRow]`.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/19394
**[Test build #82525 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/82525/testReport)** for PR 19394 at commit [`5b92fe2`](https://github.com/apache/spark/commit/5b92fe252530f4c8dab4d41772011e86ac308f80).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/19394
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/19394
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/82338/
Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #19394: [SPARK-22170][SQL] Reduce memory consumption in b...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/19394#discussion_r143321349
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/SparkPlan.scala ---
@@ -274,19 +274,26 @@ abstract class SparkPlan extends QueryPlan[SparkPlan] with Logging with Serializ
val byteArrayRdd = getByteArrayRdd()
val results = ArrayBuffer[InternalRow]()
- byteArrayRdd.collect().foreach { bytes =>
- decodeUnsafeRows(bytes).foreach(results.+=)
+ byteArrayRdd.collect().foreach { rdd =>
+ decodeUnsafeRows(rdd._2).foreach(results.+=)
}
results.toArray
}
+ private[spark] def executeCollectIterator(): (Long, Iterator[InternalRow]) = {
+ val countsAndBytes = getByteArrayRdd().collect()
+ val total = countsAndBytes.map(_._1).sum
+ val rows = countsAndBytes.iterator.flatMap(rdd => decodeUnsafeRows(rdd._2))
+ (total, rows)
+ }
+
/**
* Runs this query returning the result as an iterator of InternalRow.
*
* @note Triggers multiple jobs (one for each partition).
*/
def executeToIterator(): Iterator[InternalRow] = {
- getByteArrayRdd().toLocalIterator.flatMap(decodeUnsafeRows)
+ getByteArrayRdd().toLocalIterator.flatMap(rdd => decodeUnsafeRows(rdd._2))
--- End diff --
We don't need to collect the counts back to driver. Besides it should not be a rdd but bytes in the `flatMap`.
Maybe:
```scala
getByteArrayRdd().map(_._2).toLocalIterator.flatMap(decodeUnsafeRows(_))
```
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #19394: [SPARK-22170][SQL] Reduce memory consumption in b...
Posted by rdblue <gi...@git.apache.org>.
Github user rdblue commented on a diff in the pull request:
https://github.com/apache/spark/pull/19394#discussion_r143226714
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/BroadcastExchangeExec.scala ---
@@ -73,25 +73,37 @@ case class BroadcastExchangeExec(
try {
val beforeCollect = System.nanoTime()
// Note that we use .executeCollect() because we don't want to convert data to Scala types
- val input: Array[InternalRow] = child.executeCollect()
- if (input.length >= 512000000) {
+ val (numRows, input) = child.executeCollectIterator()
+ if (numRows >= 512000000) {
throw new SparkException(
- s"Cannot broadcast the table with more than 512 millions rows: ${input.length} rows")
+ s"Cannot broadcast the table with more than 512 millions rows: $numRows rows")
}
+
val beforeBuild = System.nanoTime()
longMetric("collectTime") += (beforeBuild - beforeCollect) / 1000000
- val dataSize = input.map(_.asInstanceOf[UnsafeRow].getSizeInBytes.toLong).sum
+
+ // Construct the relation.
+ val relation = mode.transform(input, Some(numRows))
+
+ val dataSize = relation match {
+ case map: HashedRelation =>
+ map.estimatedSize
+ case arr: Array[InternalRow] =>
+ arr.map(_.asInstanceOf[UnsafeRow].getSizeInBytes.toLong).sum
+ case _ =>
+ numRows * 512 // guess: each row is about 512 bytes
--- End diff --
The won't cause a regression because all the broadcast modes return either Array or HashedRelation. This is just in case there is a path that returns something different in the future. Maybe it would be better to throw an exception here. What do you think?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/19394
**[Test build #82486 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/82486/testReport)** for PR 19394 at commit [`3b43e11`](https://github.com/apache/spark/commit/3b43e112d11051c6ce88490530a3dbf4327efad3).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/19394
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/19394
**[Test build #82525 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/82525/testReport)** for PR 19394 at commit [`5b92fe2`](https://github.com/apache/spark/commit/5b92fe252530f4c8dab4d41772011e86ac308f80).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/19394
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/82555/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #19394: [SPARK-22170][SQL] Reduce memory consumption in b...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/spark/pull/19394
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/19394
**[Test build #82556 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/82556/testReport)** for PR 19394 at commit [`1a813ac`](https://github.com/apache/spark/commit/1a813acecb61933544c7c6ca1ebd58d5b95b7fa5).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #19394: [SPARK-22170][SQL] Reduce memory consumption in b...
Posted by rdblue <gi...@git.apache.org>.
Github user rdblue commented on a diff in the pull request:
https://github.com/apache/spark/pull/19394#discussion_r143060736
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/execution/metric/SQLMetricsSuite.scala ---
@@ -228,7 +228,7 @@ class SQLMetricsSuite extends SparkFunSuite with SQLMetricsTestUtils with Shared
testSparkPlanMetrics(df, 2, Map(
1L -> (("BroadcastHashJoin", Map(
"number of output rows" -> 2L,
- "avg hash probe (min, med, max)" -> "\n(1, 1, 1)"))))
+ "avg hash probe (min, med, max)" -> "\n(1.1, 1.1, 1.1)"))))
--- End diff --
This test was incorrect. I debugged the metric and the value is consistently 1.1, with 9 lookups and 10 probes. 10/9 = 1.1... The reason why the test was passing before is that the metrics check was hitting a [race condition](https://github.com/apache/spark/blob/master/sql/core/src/test/scala/org/apache/spark/sql/execution/metric/SQLMetricsTestUtils.scala#L165-L169) that caused the test to pass. If I change the values in master to 1.1, the tests still pass and demonstrate that the check isn't really happening.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/19394
Merged build finished. Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/19394
**[Test build #82486 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/82486/testReport)** for PR 19394 at commit [`3b43e11`](https://github.com/apache/spark/commit/3b43e112d11051c6ce88490530a3dbf4327efad3).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #19394: [SPARK-22170][SQL] Reduce memory consumption in b...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on a diff in the pull request:
https://github.com/apache/spark/pull/19394#discussion_r143315127
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/SparkPlan.scala ---
@@ -280,13 +280,20 @@ abstract class SparkPlan extends QueryPlan[SparkPlan] with Logging with Serializ
results.toArray
}
+ private[spark] def executeCollectIterator(): (Long, Iterator[InternalRow]) = {
+ val countsAndBytes = getByteArrayRdd().collect()
+ val total = countsAndBytes.map(_._1).sum
+ val rows = countsAndBytes.iterator.map(_._2).flatMap(decodeUnsafeRows)
--- End diff --
-> `flatMap(o => decodeUnsafeRows(o._2))`
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on the issue:
https://github.com/apache/spark/pull/19394
What's the other value?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/19394
**[Test build #82558 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/82558/testReport)** for PR 19394 at commit [`ab2e3ca`](https://github.com/apache/spark/commit/ab2e3cabebc7584e3803cffc727aad5b8cd16122).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/19394
**[Test build #82338 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/82338/testReport)** for PR 19394 at commit [`d82ffca`](https://github.com/apache/spark/commit/d82ffca19c84589dfcb87a81996fb3ce2e31e91a).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/19394
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/82525/
Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/19394
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/82558/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/19394
**[Test build #82532 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/82532/testReport)** for PR 19394 at commit [`56089f5`](https://github.com/apache/spark/commit/56089f5ba65f1d7d9e11b76673bcde3df37cd240).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/19394
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/82532/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/19394
**[Test build #82532 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/82532/testReport)** for PR 19394 at commit [`56089f5`](https://github.com/apache/spark/commit/56089f5ba65f1d7d9e11b76673bcde3df37cd240).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on the issue:
https://github.com/apache/spark/pull/19394
retest this please
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #19394: [SPARK-22170][SQL] Reduce memory consumption in b...
Posted by liufengdb <gi...@git.apache.org>.
Github user liufengdb commented on a diff in the pull request:
https://github.com/apache/spark/pull/19394#discussion_r142290483
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/SparkPlan.scala ---
@@ -280,13 +280,20 @@ abstract class SparkPlan extends QueryPlan[SparkPlan] with Logging with Serializ
results.toArray
}
+ private[spark] def executeCollectIterator(): (Long, Iterator[InternalRow]) = {
+ val countsAndBytes = getByteArrayRdd().collect()
--- End diff --
This still fetches all the compressed rows to the driver, before building the hashed relation. Ideally, you should fetch the rows from executors incrementally.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/19394
**[Test build #82530 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/82530/testReport)** for PR 19394 at commit [`5b92fe2`](https://github.com/apache/spark/commit/5b92fe252530f4c8dab4d41772011e86ac308f80).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #19394: [SPARK-22170][SQL] Reduce memory consumption in b...
Posted by rdblue <gi...@git.apache.org>.
Github user rdblue closed the pull request at:
https://github.com/apache/spark/pull/19394
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/19394
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/82556/
Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/19394
**[Test build #82558 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/82558/testReport)** for PR 19394 at commit [`ab2e3ca`](https://github.com/apache/spark/commit/ab2e3cabebc7584e3803cffc727aad5b8cd16122).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #19394: [SPARK-22170][SQL] Reduce memory consumption in b...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/19394#discussion_r143321383
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/SparkPlan.scala ---
@@ -274,19 +274,26 @@ abstract class SparkPlan extends QueryPlan[SparkPlan] with Logging with Serializ
val byteArrayRdd = getByteArrayRdd()
val results = ArrayBuffer[InternalRow]()
- byteArrayRdd.collect().foreach { bytes =>
- decodeUnsafeRows(bytes).foreach(results.+=)
+ byteArrayRdd.collect().foreach { rdd =>
+ decodeUnsafeRows(rdd._2).foreach(results.+=)
}
results.toArray
}
+ private[spark] def executeCollectIterator(): (Long, Iterator[InternalRow]) = {
+ val countsAndBytes = getByteArrayRdd().collect()
+ val total = countsAndBytes.map(_._1).sum
+ val rows = countsAndBytes.iterator.flatMap(rdd => decodeUnsafeRows(rdd._2))
--- End diff --
nit: `rdd` maybe confusing.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/19394
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/82486/
Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/19394
**[Test build #82338 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/82338/testReport)** for PR 19394 at commit [`d82ffca`](https://github.com/apache/spark/commit/d82ffca19c84589dfcb87a81996fb3ce2e31e91a).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/19394
Merged build finished. Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on the issue:
https://github.com/apache/spark/pull/19394
LGTM
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by rdblue <gi...@git.apache.org>.
Github user rdblue commented on the issue:
https://github.com/apache/spark/pull/19394
Yes, we've been running this in production for a few weeks now.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by rdblue <gi...@git.apache.org>.
Github user rdblue commented on the issue:
https://github.com/apache/spark/pull/19394
Ideally, this would also use a TaskMemoryManager so the driver can spill results to disk instead of dying with an OOM. Is there any plan to add a memory manager for the driver?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/19394
**[Test build #82524 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/82524/testReport)** for PR 19394 at commit [`f6613c2`](https://github.com/apache/spark/commit/f6613c2651691ca149390ea2407856ba390590cb).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/19394
Merged build finished. Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #19394: [SPARK-22170][SQL] Reduce memory consumption in b...
Posted by rdblue <gi...@git.apache.org>.
Github user rdblue commented on a diff in the pull request:
https://github.com/apache/spark/pull/19394#discussion_r143228054
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/BroadcastExchangeExec.scala ---
@@ -73,25 +73,37 @@ case class BroadcastExchangeExec(
try {
val beforeCollect = System.nanoTime()
// Note that we use .executeCollect() because we don't want to convert data to Scala types
- val input: Array[InternalRow] = child.executeCollect()
- if (input.length >= 512000000) {
+ val (numRows, input) = child.executeCollectIterator()
+ if (numRows >= 512000000) {
throw new SparkException(
- s"Cannot broadcast the table with more than 512 millions rows: ${input.length} rows")
+ s"Cannot broadcast the table with more than 512 millions rows: $numRows rows")
}
+
val beforeBuild = System.nanoTime()
longMetric("collectTime") += (beforeBuild - beforeCollect) / 1000000
- val dataSize = input.map(_.asInstanceOf[UnsafeRow].getSizeInBytes.toLong).sum
+
+ // Construct the relation.
+ val relation = mode.transform(input, Some(numRows))
+
+ val dataSize = relation match {
+ case map: HashedRelation =>
+ map.estimatedSize
--- End diff --
True, but I think using the accurate size is better. Maybe we should revisit the 8gb maximum instead of using a size that isn't correct. What's the reason for the 8gb max?
In master, building a broadcast table just under the 8gb limit would require 8gb for the rows (what we actually count here), plus the size of those rows compressed, which is probably at least 1gb. So we're talking about a pretty huge broadcast table to hit the case where this check will cause failures that didn't happen before. Not that it couldn't happen, but I've never seen this in practice and we have some huge jobs.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by rdblue <gi...@git.apache.org>.
Github user rdblue commented on the issue:
https://github.com/apache/spark/pull/19394
Here's the error message: TestFailedException: 347.5272 was not greater than 1000
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #19394: [SPARK-22170][SQL] Reduce memory consumption in b...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/19394#discussion_r143207681
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/BroadcastExchangeExec.scala ---
@@ -73,25 +73,37 @@ case class BroadcastExchangeExec(
try {
val beforeCollect = System.nanoTime()
// Note that we use .executeCollect() because we don't want to convert data to Scala types
- val input: Array[InternalRow] = child.executeCollect()
- if (input.length >= 512000000) {
+ val (numRows, input) = child.executeCollectIterator()
+ if (numRows >= 512000000) {
throw new SparkException(
- s"Cannot broadcast the table with more than 512 millions rows: ${input.length} rows")
+ s"Cannot broadcast the table with more than 512 millions rows: $numRows rows")
}
+
val beforeBuild = System.nanoTime()
longMetric("collectTime") += (beforeBuild - beforeCollect) / 1000000
- val dataSize = input.map(_.asInstanceOf[UnsafeRow].getSizeInBytes.toLong).sum
+
+ // Construct the relation.
+ val relation = mode.transform(input, Some(numRows))
+
+ val dataSize = relation match {
+ case map: HashedRelation =>
+ map.estimatedSize
+ case arr: Array[InternalRow] =>
--- End diff --
Yeah, right. `BroadcastNestedLoopJoinExec` uses `IdentityBroadcastMode`.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #19394: [SPARK-22170][SQL] Reduce memory consumption in b...
Posted by rdblue <gi...@git.apache.org>.
Github user rdblue commented on a diff in the pull request:
https://github.com/apache/spark/pull/19394#discussion_r143285737
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/ConfigBehaviorSuite.scala ---
@@ -58,7 +58,7 @@ class ConfigBehaviorSuite extends QueryTest with SharedSQLContext {
withSQLConf(SQLConf.RANGE_EXCHANGE_SAMPLE_SIZE_PER_PARTITION.key -> "1") {
// If we only sample one point, the range boundaries will be pretty bad and the
// chi-sq value would be very high.
- assert(computeChiSquareTest() > 1000)
+ assert(computeChiSquareTest() > 300)
--- End diff --
@rxin, the difference that is causing this to fail is that `rdd.id` is 15 instead of 14 because of the change to `getByteArrayRdd`. That id is used to seed the random generator and the chi-sq value isn't high enough. The previous value must have been unusually high. With so few data points, this can vary quite a bit so I think changing the bound to 300 is a good fix.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/19394
**[Test build #82530 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/82530/testReport)** for PR 19394 at commit [`5b92fe2`](https://github.com/apache/spark/commit/5b92fe252530f4c8dab4d41772011e86ac308f80).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/19394
**[Test build #82524 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/82524/testReport)** for PR 19394 at commit [`f6613c2`](https://github.com/apache/spark/commit/f6613c2651691ca149390ea2407856ba390590cb).
* This patch **fails PySpark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #19394: [SPARK-22170][SQL] Reduce memory consumption in b...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/19394#discussion_r143093875
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/BroadcastExchangeExec.scala ---
@@ -73,25 +73,37 @@ case class BroadcastExchangeExec(
try {
val beforeCollect = System.nanoTime()
// Note that we use .executeCollect() because we don't want to convert data to Scala types
- val input: Array[InternalRow] = child.executeCollect()
- if (input.length >= 512000000) {
+ val (numRows, input) = child.executeCollectIterator()
+ if (numRows >= 512000000) {
throw new SparkException(
- s"Cannot broadcast the table with more than 512 millions rows: ${input.length} rows")
+ s"Cannot broadcast the table with more than 512 millions rows: $numRows rows")
}
+
val beforeBuild = System.nanoTime()
longMetric("collectTime") += (beforeBuild - beforeCollect) / 1000000
- val dataSize = input.map(_.asInstanceOf[UnsafeRow].getSizeInBytes.toLong).sum
+
+ // Construct the relation.
+ val relation = mode.transform(input, Some(numRows))
+
+ val dataSize = relation match {
+ case map: HashedRelation =>
+ map.estimatedSize
--- End diff --
The estimated size of hashed relation maybe larger than the accurate data size calculated before. I guess this may be regression under some cases.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/19394
**[Test build #82555 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/82555/testReport)** for PR 19394 at commit [`a2976fe`](https://github.com/apache/spark/commit/a2976fe093f0b0afe9a8b01a876d401fb0bf2d04).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on the issue:
https://github.com/apache/spark/pull/19394
This is a pretty cool fix
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on the issue:
https://github.com/apache/spark/pull/19394
cc @liufengdb
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/19394
**[Test build #82555 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/82555/testReport)** for PR 19394 at commit [`a2976fe`](https://github.com/apache/spark/commit/a2976fe093f0b0afe9a8b01a876d401fb0bf2d04).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by rdblue <gi...@git.apache.org>.
Github user rdblue commented on the issue:
https://github.com/apache/spark/pull/19394
@rxin, any idea why this would fail ConfigBehaviorSuite? I don't think the failure is related because that test doesn't use a broadcast join. Should I rebase on master?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on the issue:
https://github.com/apache/spark/pull/19394
Thanks! Merged to master.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #19394: [SPARK-22170][SQL] Reduce memory consumption in b...
Posted by rdblue <gi...@git.apache.org>.
Github user rdblue commented on a diff in the pull request:
https://github.com/apache/spark/pull/19394#discussion_r143286104
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/BroadcastExchangeExec.scala ---
@@ -73,25 +73,37 @@ case class BroadcastExchangeExec(
try {
val beforeCollect = System.nanoTime()
// Note that we use .executeCollect() because we don't want to convert data to Scala types
--- End diff --
Updated.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #19394: [SPARK-22170][SQL] Reduce memory consumption in b...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on a diff in the pull request:
https://github.com/apache/spark/pull/19394#discussion_r143077148
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/BroadcastExchangeExec.scala ---
@@ -73,25 +73,37 @@ case class BroadcastExchangeExec(
try {
val beforeCollect = System.nanoTime()
// Note that we use .executeCollect() because we don't want to convert data to Scala types
- val input: Array[InternalRow] = child.executeCollect()
- if (input.length >= 512000000) {
+ val (numRows, input) = child.executeCollectIterator()
+ if (numRows >= 512000000) {
throw new SparkException(
- s"Cannot broadcast the table with more than 512 millions rows: ${input.length} rows")
+ s"Cannot broadcast the table with more than 512 millions rows: $numRows rows")
}
+
val beforeBuild = System.nanoTime()
longMetric("collectTime") += (beforeBuild - beforeCollect) / 1000000
- val dataSize = input.map(_.asInstanceOf[UnsafeRow].getSizeInBytes.toLong).sum
+
+ // Construct the relation.
+ val relation = mode.transform(input, Some(numRows))
+
+ val dataSize = relation match {
+ case map: HashedRelation =>
+ map.estimatedSize
+ case arr: Array[InternalRow] =>
+ arr.map(_.asInstanceOf[UnsafeRow].getSizeInBytes.toLong).sum
+ case _ =>
+ numRows * 512 // guess: each row is about 512 bytes
--- End diff --
We did the check in line 99. Thus, this could cause a false alarm. That means, this could cause a regression, right?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/19394
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/82530/
Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #19394: [SPARK-22170][SQL] Reduce memory consumption in b...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/19394#discussion_r143093472
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/BroadcastExchangeExec.scala ---
@@ -73,25 +73,37 @@ case class BroadcastExchangeExec(
try {
val beforeCollect = System.nanoTime()
// Note that we use .executeCollect() because we don't want to convert data to Scala types
--- End diff --
This comment is obsolete.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #19394: [SPARK-22170][SQL] Reduce memory consumption in b...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on a diff in the pull request:
https://github.com/apache/spark/pull/19394#discussion_r143232991
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/BroadcastExchangeExec.scala ---
@@ -73,25 +73,37 @@ case class BroadcastExchangeExec(
try {
val beforeCollect = System.nanoTime()
// Note that we use .executeCollect() because we don't want to convert data to Scala types
- val input: Array[InternalRow] = child.executeCollect()
- if (input.length >= 512000000) {
+ val (numRows, input) = child.executeCollectIterator()
+ if (numRows >= 512000000) {
throw new SparkException(
- s"Cannot broadcast the table with more than 512 millions rows: ${input.length} rows")
+ s"Cannot broadcast the table with more than 512 millions rows: $numRows rows")
}
+
val beforeBuild = System.nanoTime()
longMetric("collectTime") += (beforeBuild - beforeCollect) / 1000000
- val dataSize = input.map(_.asInstanceOf[UnsafeRow].getSizeInBytes.toLong).sum
+
+ // Construct the relation.
+ val relation = mode.transform(input, Some(numRows))
+
+ val dataSize = relation match {
+ case map: HashedRelation =>
+ map.estimatedSize
+ case arr: Array[InternalRow] =>
+ arr.map(_.asInstanceOf[UnsafeRow].getSizeInBytes.toLong).sum
+ case _ =>
+ numRows * 512 // guess: each row is about 512 bytes
--- End diff --
Yeah, it makes more sense to issue an exception.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on the issue:
https://github.com/apache/spark/pull/19394
Is the fix being used in the Netflix production environment?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #19394: [SPARK-22170][SQL] Reduce memory consumption in b...
Posted by rdblue <gi...@git.apache.org>.
GitHub user rdblue reopened a pull request:
https://github.com/apache/spark/pull/19394
[SPARK-22170][SQL] Reduce memory consumption in broadcast joins.
## What changes were proposed in this pull request?
This updates the broadcast join code path to lazily decompress pages and
iterate through UnsafeRows to prevent all rows from being held in memory
while the broadcast table is being built.
## How was this patch tested?
Existing tests.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/rdblue/spark broadcast-driver-memory
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/19394.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #19394
----
----
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/19394
Merged build finished. Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by rdblue <gi...@git.apache.org>.
Github user rdblue commented on the issue:
https://github.com/apache/spark/pull/19394
Anyone have a clue what the python error could be? It doesn't look related.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #19394: [SPARK-22170][SQL] Reduce memory consumption in b...
Posted by rdblue <gi...@git.apache.org>.
Github user rdblue commented on a diff in the pull request:
https://github.com/apache/spark/pull/19394#discussion_r143226823
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/BroadcastExchangeExec.scala ---
@@ -73,25 +73,37 @@ case class BroadcastExchangeExec(
try {
val beforeCollect = System.nanoTime()
// Note that we use .executeCollect() because we don't want to convert data to Scala types
- val input: Array[InternalRow] = child.executeCollect()
- if (input.length >= 512000000) {
+ val (numRows, input) = child.executeCollectIterator()
+ if (numRows >= 512000000) {
throw new SparkException(
- s"Cannot broadcast the table with more than 512 millions rows: ${input.length} rows")
+ s"Cannot broadcast the table with more than 512 millions rows: $numRows rows")
}
+
val beforeBuild = System.nanoTime()
longMetric("collectTime") += (beforeBuild - beforeCollect) / 1000000
- val dataSize = input.map(_.asInstanceOf[UnsafeRow].getSizeInBytes.toLong).sum
+
+ // Construct the relation.
+ val relation = mode.transform(input, Some(numRows))
+
+ val dataSize = relation match {
+ case map: HashedRelation =>
+ map.estimatedSize
+ case arr: Array[InternalRow] =>
--- End diff --
Yeah, both cases are used. The only one that won't match is `case _ => ...`
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #19394: [SPARK-22170][SQL] Reduce memory consumption in b...
Posted by rdblue <gi...@git.apache.org>.
Github user rdblue commented on a diff in the pull request:
https://github.com/apache/spark/pull/19394#discussion_r143286130
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/BroadcastExchangeExec.scala ---
@@ -73,25 +73,37 @@ case class BroadcastExchangeExec(
try {
val beforeCollect = System.nanoTime()
// Note that we use .executeCollect() because we don't want to convert data to Scala types
- val input: Array[InternalRow] = child.executeCollect()
- if (input.length >= 512000000) {
+ val (numRows, input) = child.executeCollectIterator()
+ if (numRows >= 512000000) {
throw new SparkException(
- s"Cannot broadcast the table with more than 512 millions rows: ${input.length} rows")
+ s"Cannot broadcast the table with more than 512 millions rows: $numRows rows")
}
+
val beforeBuild = System.nanoTime()
longMetric("collectTime") += (beforeBuild - beforeCollect) / 1000000
- val dataSize = input.map(_.asInstanceOf[UnsafeRow].getSizeInBytes.toLong).sum
+
+ // Construct the relation.
+ val relation = mode.transform(input, Some(numRows))
+
+ val dataSize = relation match {
+ case map: HashedRelation =>
+ map.estimatedSize
+ case arr: Array[InternalRow] =>
+ arr.map(_.asInstanceOf[UnsafeRow].getSizeInBytes.toLong).sum
+ case _ =>
+ numRows * 512 // guess: each row is about 512 bytes
--- End diff --
Fixed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19394: [SPARK-22170][SQL] Reduce memory consumption in broadcas...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/19394
Merged build finished. Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #19394: [SPARK-22170][SQL] Reduce memory consumption in b...
Posted by rdblue <gi...@git.apache.org>.
Github user rdblue commented on a diff in the pull request:
https://github.com/apache/spark/pull/19394#discussion_r143317742
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/ConfigBehaviorSuite.scala ---
@@ -58,7 +58,7 @@ class ConfigBehaviorSuite extends QueryTest with SharedSQLContext {
withSQLConf(SQLConf.RANGE_EXCHANGE_SAMPLE_SIZE_PER_PARTITION.key -> "1") {
// If we only sample one point, the range boundaries will be pretty bad and the
// chi-sq value would be very high.
- assert(computeChiSquareTest() > 1000)
+ assert(computeChiSquareTest() > 300)
--- End diff --
Updating the SparkPlan code to avoid creating unnecessary RDDs fixed this test because the same random seed is used. But, I think we should keep this change so we don't have to track down this problem again in the future. This bound is perfectly safe because the balanced chi-sq value is 10, while the unbalanced is much larger.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org