awslabs / awslabs/amazon-dynamodb-tools
[bulk] The watchdog kills a run on the first OutOfMemoryError, even where Spark could have recovered
- Dominant language
- Python
- Stars
- 181
- Forks
- 50
- Avg merge
- 4h 11m
- Merged PRs (30d)
- 36
Description
## The policy today
`_watch_log_group` stops the Glue job the moment **any** stream logs a line matching
`OutOfMemoryError:` -- and it scans executor streams as well as the driver's. One executor
hitting a heap error therefore ends the whole run.
For a small job that is the right call. Measured on `--XNumberOfWorkers 2` (one executor,
four cores), from the executor's own log:
```
ERROR Executor:275 - Exception in task 0.0 in stage 2.0 (TID 200)
java.lang.OutOfMemoryError: Java heap space
ERROR Executor:267 - Executor launch task task 0.1 in stage 2.0 (TID 201) failed, reason:
Task ...TaskRunner@12270630 rejected from ...ThreadPoolExecutor@ca55f7c[Shutting down, ...]
ERROR Executor:267 - Executor launch task task 0.2 ... rejected from ...[Shutting down, ...]
ERROR Executor:267 - Executor launch task task 0.3 ... rejected from ...[Shutting down, ...]
```
All four attempts were consumed within the same second: the retries had nowhere to go
because the only executor was already shutting down. Stopping saved a couple of doomed
minutes.
## Where it may be wrong
The default is **220 workers**. An executor that dies there takes its tasks with it, but
Spark re-schedules them on the ~219 others, and `spark.task.maxFailures` allows four
attempts. A single skewed partition, or one executor unlucky enough to hold two large tasks
at once, can produce a heap error the run would have recovered from.
We kill it anyway -- including a `copy` or `fill` that is hours in. That trade (one OOM line
= abort) has never been measured at scale, and it is not obviously the right one.
## Worth measuring before changing
- Does a 220-worker run actually recover from a single executor OOM, or does the same
aggregation just OOM wherever it lands? (For a whole-table sort or `collect_list` the
answer is probably "dies everywhere", since the task is the same size on every executor.)
- Is a *driver* OOM ever recoverable? Almost certainly not -- worth keeping the immediate
stop for that one regardless.
## Possible shapes
- Distinguish driver from executor: stop immediately for the driver, and for an executor
only after N distinct occurrences or once Spark reports the stage aborted (`Task N in
stage M failed 4 times; aborting job` is logged, and is the point where recovery has
provably failed).
- Or keep the current behaviour and simply say so in the output -- "stopped after the first
executor ran out of memory; re-run with a larger --XWorkerType" -- which #345 now does.
Filed as a question, not a bug: the current behaviour is defensible, but it should be
deliberate rather than a side effect of matching one string in every stream.
Contributor guide
Research direction
Start at `_watch_log_group`, which currently matches OutOfMemoryError lines across driver and executor streams. Measure the described 220-worker recovery case and compare it with driver OOM and Spark stage-abort messages; done means a documented policy decision supported by those observations, whether the current behavior is retained or changed.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- aws, python, spark
- Domain
- distributed-systems, observability-sre
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Active
- Clarity
- Needs clarification
- Newbie friendliness
- 30/100