Ivan Sadikov created SPARK-61322:
------------------------------------
Summary: Fail the stage with a clear error when a serialized task
exceeds the JVM array size limit
Key: SPARK-61322
URL: https://issues.apache.org/jira/browse/SPARK-61322
Project: Spark
Issue Type: Bug
Components: Spark Core
Affects Versions: 4.3.0
Reporter: Ivan Sadikov
The driver serializes each task into a single byte array before launching it. A
task can carry its partition's data: for example, SparkContext.parallelize or a
local relation puts the rows of each partition inside its task. If the
serialized bytes are larger than the JVM's maximum array size (about 2 GB), the
JVM throws: java.lang.OutOfMemoryError: Required array length 2147483639 + 626
is too large or, depending on the JDK: java.lang.OutOfMemoryError: Requested
array size exceeds VM limit
This isn't heap exhaustion. It's a fixed limit, so retrying can never succeed.
The scheduler handles it badly in two places:
1. Task serialization (TaskSetManager.prepareLaunchingTask). Serialization
failures are caught with case NonFatal(e), but OutOfMemoryError is fatal, so it
is
not caught. The task set is not aborted. The error escapes the scheduler's
resource-offer path and is rethrown from the driver RPC message loop
(Inbox.dealWithFatalError). Depending on the uncaught-exception handler,
either the driver dies or the job never finishes. Either way, one oversized
query
can affect the whole application.
2. Task binary serialization (DAGScheduler.submitMissingTasks). The per-stage
task binary ((rdd, func) / (rdd, shuffleDep)) is caught by case e: Throwable.
The stage is aborted with Task serialization failed:
java.lang.OutOfMemoryError: ..., with the OutOfMemoryError as the cause. The
message doesn't help the
user, and tools that classify failures by cause type report it as an
out-of-memory condition. That suggests raising driver memory, which cannot help.
Reproduction (needs a driver heap large enough to hold the data, e.g.
--driver-memory 16g):
{code}
// ~3 GB of driver-local data in a single partition, shipped inside one task.
sc.parallelize(Seq.fill(3)(new Array[Byte](1 << 30)), 1).count()
{code}
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
