[SPARK-59407][UI][HISTORY] Improve compressed line handling in Spark History server - #58700
[SPARK-59407][UI][HISTORY] Improve compressed line handling in Spark History server#58700holdenk wants to merge 5 commits into
Conversation
Replay materializes each event log line as a String with no length bound, so a corrupt or unexpectedly large log can exhaust the memory of the process replaying it. Cap lines at spark.history.fs.eventLog.maxLineLength (default 512m, <= 0 disables): longer lines are drained and skipped with a warning while the remaining events still replay. Co-authored-by: Cursor <cursoragent@cursor.com> Co-Authored-By: Holden Karau <holden@pigscanfly.ca>
…config Every new config entry needs one. A History Server replay setting cannot change how a view or UDF body resolves, so NOT_APPLICABLE. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Co-Authored-By: Holden Karau <holden@pigscanfly.ca>
…y ships in The config was marked 4.4.0, which is only true if this lands on the 4.x line and nowhere else. It is being backported to the maintenance branches, so the earliest release a user can get it in is 4.1.4, and that is what the version field is for -- 3.5 and 4.0 name their own next patch, everything from 4.1 up names 4.1.4. Same shape as the UDT loading configs, which read 4.1.3 on master for the same reason. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Co-Authored-By: Holden Karau <holden@pigscanfly.ca>
…rsion A single version tells an operator nothing about a backported config: it does not say which maintenance releases have it. Name the whole set in the doc, and declare the version this branch actually first ships in. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Co-Authored-By: Holden Karau <holden@pigscanfly.ca>
The version note only reached the config doc, so nobody reading the docs tables saw it. Add the row. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Co-Authored-By: Holden Karau <holden@pigscanfly.ca>
| .version("4.3.0") | ||
| .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE) | ||
| .bytesConf(ByteUnit.BYTE) | ||
| .createWithDefaultString("512m") |
There was a problem hiding this comment.
Can we make this as createOptional since we plan to backport to old branches?
HyukjinKwon
left a comment
HyukjinKwon
left a comment
There was a problem hiding this comment.
0 blocking, 1 non-blocking, 1 nit.
A well-scoped, backward-compatible memory-safety improvement for History-server replay; the only items are two minor documentation/clarity points.
Suggestions (1)
- core/src/main/scala/org/apache/spark/internal/config/History.scala:179: Config is declared in bytes but enforced as a UTF-16 character count; document that it counts characters. -- see inline
Nits: 1 minor item (see inline comments).
Verification
I traced the new reader against the replaced Source.getLines(). In-bound lines replay identically: same \n splitting, CRLF stripping, newline-less final line, and malformed-UTF-8 failure (CodingErrorAction.REPORT). Memory is bounded — the only input-scaled allocation is the per-line StringBuilder, capped by sb.length() < maxLineLength, and over-long lines are drained without materialization; the BufferedReader/InputStreamReader internal buffers are fixed-size. A MalformedInputException still surfaces through the lazy iterator into replay()'s existing IOException rethrow. maxLineLength(conf) maps <= 0 and > Int.MaxValue to Int.MaxValue, so there is no overflow and 0/negative cleanly disables the cap.
|
|
||
| /** | ||
| * Reads '\n'-terminated lines like Source.getLines(), but never materializes more than | ||
| * [[maxLineLength]] characters of a single line. An over-long line is drained and skipped |
There was a problem hiding this comment.
A [[...]] wiki-link only resolves to documented members, and maxLineLength is a non-val constructor parameter, which Scaladoc doesn't document — so this renders as literal text rather than a link.
| * [[maxLineLength]] characters of a single line. An over-long line is drained and skipped | |
| * `maxLineLength` characters of a single line. An over-long line is drained and skipped |
| .version("4.3.0") | ||
| .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE) | ||
| .bytesConf(ByteUnit.BYTE) | ||
| .createWithDefaultString("512m") |
There was a problem hiding this comment.
This is declared as a byte size (bytesConf, 512m), but the cap is enforced as a UTF-16 character count downstream — sb.length() < maxLineLength in ReplayListenerBus.boundedLines. So for multi-byte UTF-8 the effective byte threshold is larger than configured, and a retained line can hold up to ~2x the configured number of bytes in heap (Java char is 2 bytes).
The class parameter doc already says "characters"; worth saying the same in this entry's .doc(...) and in the docs/monitoring.md row so the documented contract matches enforcement.
There was a problem hiding this comment.
Can we actually make this conf optional too? Since we're backporting them to old branches.
3 participants
What changes were proposed in this pull request?
Skip excessively long compressed lines.
Why are the changes needed?
Individual long compressed lines can cause slowness loading a previous Spark job.
Does this PR introduce any user-facing change?
New configurable param
How was this patch tested?
New test
Was this patch authored or co-authored using generative AI tooling?
Claude and cursor