Josh Rosen created SPARK-39771:
----------------------------------
Summary: If spark.default.parallelism is unset, RDD
defaultPartitioner may pick a value that is too large to successfully run
Key: SPARK-39771
URL: https://issues.apache.org/jira/browse/SPARK-39771
Project: Spark
Issue Type: Improvement
Components: Spark Core
Affects Versions: 1.0.0
Reporter: Josh Rosen
[According to its
docs|https://github.com/apache/spark/blob/899f6c90eb2de5b46a36710a131d7417010ce4b3/core/src/main/scala/org/apache/spark/Partitioner.scala#L45-L65],
{{Partitioner.defaultPartitioner}} will use the maximum number of RDD
partitions as its partition count when {{spark.default.parallelism}} is not
set. If that number of upstream partitions is very large then this can result
in shuffles where {{{}numMappers * numReducers = numMappers^2{}}}, which can
cause various problems that prevent the job from successfully running.
To help users identify when they have run into this problem, I think we should
add warning logs to Spark.
As an example of the problem, let's say that I have an RDD with 100,000
partitions and then do a {{reduceByKey}} on it without specifying an explicit
partitioner or partition count. In this case, Spark will plan a reduce stage
with 100,000 partitions:
{code:java}
scala> sc.parallelize(1 to 100000, 100000).map(x => (x, x)).reduceByKey(_ +
_).toDebugString
res7: String =
(100000) ShuffledRDD[21] at reduceByKey at <console>:25 []
+-(100000) MapPartitionsRDD[20] at map at <console>:25 []
| ParallelCollectionRDD[19] at parallelize at <console>:25 []
{code}
This results in the creation of 10 billion shuffle blocks, so if this job
_does_ run it is likely to be extremely show. However, it's more likely that
the driver will crash when serializing map output statuses: if we were able to
use one bit per mapper / reducer pair (which is probably overly optimistic in
terms of compressibility) then the map statuses would be ~1.25 gigabytes!
I don't think that users are likely to intentionally wind up in this scenario:
it's more likely that either (a) their job depends on
{{spark.default.parallelism}} being set but it was run on an environment
lacking a value for that config, or (b) their input data significantly grew in
size. These scenarios may be rare, but they can be frustrating to debug
(especially if a failure occurs midway through a long-running job).
I think we should do something to handle this scenario.
A good starting point might be for {{Partitioner.defaultPartitioner}} to log a
warning when the default partition size exceeds some threshold.
In addition, I think it might be a good idea to log a similar warning in
{{MapOutputTrackerMaster}} right before we start trying to serialize map
statuses: in a real-world situation where this problem cropped up, the map
stage ran successfully but the driver crashed when serializing map statuses.
Putting a warning about partition counts here makes it more likely that users
will spot that error in the logs and be able to identify the source of the
problem (compared to a warning that appears much earlier in the job and
therefore much farther from the likely site of a crash).
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org
For additional commands, e-mail: issues-h...@spark.apache.org
