Skip to content

Commit ba5bcad

Browse files
ash211mateiz
authored andcommitted
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 <[email protected]> Closes apache#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
1 parent 7ff8c45 commit ba5bcad

File tree

2 files changed

+7
-8
lines changed

2 files changed

+7
-8
lines changed

core/src/main/scala/org/apache/spark/rdd/RDD.scala

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1064,11 +1064,10 @@ abstract class RDD[T: ClassTag](
10641064
// greater than totalParts because we actually cap it at totalParts in runJob.
10651065
var numPartsToTry = 1
10661066
if (partsScanned > 0) {
1067-
// If we didn't find any rows after the first iteration, just try all partitions next.
1068-
// Otherwise, interpolate the number of partitions we need to try, but overestimate it
1069-
// by 50%.
1067+
// If we didn't find any rows after the previous iteration, quadruple and retry. Otherwise,
1068+
// interpolate the number of partitions we need to try, but overestimate it by 50%.
10701069
if (buf.size == 0) {
1071-
numPartsToTry = totalParts - 1
1070+
numPartsToTry = partsScanned * 4
10721071
} else {
10731072
numPartsToTry = (1.5 * num * partsScanned / buf.size).toInt
10741073
}

python/pyspark/rdd.py

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1089,11 +1089,11 @@ def take(self, num):
10891089
# we actually cap it at totalParts in runJob.
10901090
numPartsToTry = 1
10911091
if partsScanned > 0:
1092-
# If we didn't find any rows after the first iteration, just
1093-
# try all partitions next. Otherwise, interpolate the number
1094-
# of partitions we need to try, but overestimate it by 50%.
1092+
# If we didn't find any rows after the previous iteration,
1093+
# quadruple and retry. Otherwise, interpolate the number of
1094+
# partitions we need to try, but overestimate it by 50%.
10951095
if len(items) == 0:
1096-
numPartsToTry = totalParts - 1
1096+
numPartsToTry = partsScanned * 4
10971097
else:
10981098
numPartsToTry = int(1.5 * num * partsScanned / len(items))
10991099

0 commit comments

Comments
 (0)