diff options
author | Andrew Ash <andrew@andrewash.com> | 2014-09-05 18:52:05 -0700 |
---|---|---|
committer | Matei Zaharia <matei@databricks.com> | 2014-09-05 18:52:18 -0700 |
commit | 28ce67b5d3b0c12df3d158f042faa60aece89628 (patch) | |
tree | 8ef42a82a81fb73c07f87bc9b60718d5f93347a1 | |
parent | 6b128be0328bb84b4c477e3365a6d37cb778b523 (diff) | |
download | spark-28ce67b5d3b0c12df3d158f042faa60aece89628.tar.gz spark-28ce67b5d3b0c12df3d158f042faa60aece89628.tar.bz2 spark-28ce67b5d3b0c12df3d158f042faa60aece89628.zip |
SPARK-3211 .take() is OOM-prone with empty partitions
Instead of jumping straight from 1 partition to all partitions, do exponential
growth and double the number of partitions to attempt each time instead.
Fix proposed by Paul Nepywoda
Author: Andrew Ash <andrew@andrewash.com>
Closes #2117 from ash211/SPARK-3211 and squashes the following commits:
8b2299a [Andrew Ash] Quadruple instead of double for a minor speedup
e5f7e4d [Andrew Ash] Update comment to better reflect what we're doing
09a27f7 [Andrew Ash] Update PySpark to be less OOM-prone as well
3a156b8 [Andrew Ash] SPARK-3211 .take() is OOM-prone with empty partitions
(cherry picked from commit ba5bcaddecd54811d45c5fc79a013b3857d4c633)
Signed-off-by: Matei Zaharia <matei@databricks.com>
-rw-r--r-- | core/src/main/scala/org/apache/spark/rdd/RDD.scala | 7 | ||||
-rw-r--r-- | python/pyspark/rdd.py | 8 |
2 files changed, 7 insertions, 8 deletions
diff --git a/core/src/main/scala/org/apache/spark/rdd/RDD.scala b/core/src/main/scala/org/apache/spark/rdd/RDD.scala index 0159003c88..006fad8db2 100644 --- a/core/src/main/scala/org/apache/spark/rdd/RDD.scala +++ b/core/src/main/scala/org/apache/spark/rdd/RDD.scala @@ -1064,11 +1064,10 @@ abstract class RDD[T: ClassTag]( // greater than totalParts because we actually cap it at totalParts in runJob. var numPartsToTry = 1 if (partsScanned > 0) { - // If we didn't find any rows after the first iteration, just try all partitions next. - // Otherwise, interpolate the number of partitions we need to try, but overestimate it - // by 50%. + // If we didn't find any rows after the previous iteration, quadruple and retry. Otherwise, + // interpolate the number of partitions we need to try, but overestimate it by 50%. if (buf.size == 0) { - numPartsToTry = totalParts - 1 + numPartsToTry = partsScanned * 4 } else { numPartsToTry = (1.5 * num * partsScanned / buf.size).toInt } diff --git a/python/pyspark/rdd.py b/python/pyspark/rdd.py index bfefc26123..2b47b6c18e 100644 --- a/python/pyspark/rdd.py +++ b/python/pyspark/rdd.py @@ -1132,11 +1132,11 @@ class RDD(object): # we actually cap it at totalParts in runJob. numPartsToTry = 1 if partsScanned > 0: - # If we didn't find any rows after the first iteration, just - # try all partitions next. Otherwise, interpolate the number - # of partitions we need to try, but overestimate it by 50%. + # If we didn't find any rows after the previous iteration, + # quadruple and retry. Otherwise, interpolate the number of + # partitions we need to try, but overestimate it by 50%. if len(items) == 0: - numPartsToTry = totalParts - 1 + numPartsToTry = partsScanned * 4 else: numPartsToTry = int(1.5 * num * partsScanned / len(items)) |