You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ho...@apache.org on 2026/09/24 23:32:16 UTC
(spark) branch master updated: [SPARK-59407][UI] Improve compressed line handling in Spark History server
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 38258cbea90c [SPARK-59407][UI] Improve compressed line handling in Spark History server
38258cbea90c is described below
commit 38258cbea90c7540d71db86a64cfd18ccc0dffec
Author: Holden Karau <ho...@pigscanfly.ca>
AuthorDate: Thu Sep 24 16:32:05 2026 -0700
[SPARK-59407][UI] Improve compressed line handling in Spark History server
### 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
Closes #58700 from holdenk/f040-compressed-file-handling.
Lead-authored-by: Holden Karau <ho...@pigscanfly.ca>
Co-authored-by: Holden Karau <ho...@snowflake.com>
Signed-off-by: Holden Karau <ho...@snowflake.com>
---
.../deploy/history/EventLogFileCompactor.scala | 2 +-
.../spark/deploy/history/FsHistoryProvider.scala | 4 +-
.../org/apache/spark/internal/config/History.scala | 13 +++
.../apache/spark/scheduler/ReplayListenerBus.scala | 100 ++++++++++++++++++++-
.../spark/scheduler/ReplayListenerSuite.scala | 31 +++++++
docs/monitoring.md | 13 +++
6 files changed, 156 insertions(+), 7 deletions(-)
diff --git a/core/src/main/scala/org/apache/spark/deploy/history/EventLogFileCompactor.scala b/core/src/main/scala/org/apache/spark/deploy/history/EventLogFileCompactor.scala
index e0227641fc0e..8d9be3877c78 100644
--- a/core/src/main/scala/org/apache/spark/deploy/history/EventLogFileCompactor.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/history/EventLogFileCompactor.scala
@@ -113,7 +113,7 @@ class EventLogFileCompactor(
* them via replaying events in given files.
*/
private def initializeBuilders(fs: FileSystem, files: Seq[Path]): Seq[EventFilterBuilder] = {
- val bus = new ReplayListenerBus()
+ val bus = new ReplayListenerBus(ReplayListenerBus.maxLineLength(sparkConf))
val builders = ServiceLoader.load(classOf[EventFilterBuilder],
Utils.getContextOrSparkClassLoader).asScala.toSeq
diff --git a/core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala b/core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala
index 748aec37570d..2a345c92d7a9 100644
--- a/core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala
@@ -1075,7 +1075,7 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
val shouldHalt = enableOptimizations &&
((!appCompleted && fastInProgressParsing) || reparseChunkSize > 0)
- val bus = new ReplayListenerBus()
+ val bus = new ReplayListenerBus(ReplayListenerBus.maxLineLength(conf))
val listener = new AppListingListener(reader, clock, shouldHalt, this)
bus.addListener(listener)
@@ -1427,7 +1427,7 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
// to parse the event logs in the SHS.
val replayConf = conf.clone().set(ASYNC_TRACKING_ENABLED, false)
val trackingStore = new ElementTrackingStore(store, replayConf)
- val replayBus = new ReplayListenerBus()
+ val replayBus = new ReplayListenerBus(ReplayListenerBus.maxLineLength(conf))
val listener = new AppStatusListener(trackingStore, replayConf, false,
lastUpdateTime = Some(lastUpdated))
replayBus.addListener(listener)
diff --git a/core/src/main/scala/org/apache/spark/internal/config/History.scala b/core/src/main/scala/org/apache/spark/internal/config/History.scala
index 570113d145a1..7353481b6582 100644
--- a/core/src/main/scala/org/apache/spark/internal/config/History.scala
+++ b/core/src/main/scala/org/apache/spark/internal/config/History.scala
@@ -165,6 +165,19 @@ private[spark] object History {
.bytesConf(ByteUnit.BYTE)
.createWithDefaultString("1m")
+ val EVENT_LOG_MAX_LINE_LENGTH =
+ ConfigBuilder("spark.history.fs.eventLog.maxLineLength")
+ .doc("Maximum length of a single event log line during replay. Lines longer than " +
+ "this are skipped with a warning instead of being read into memory, bounding the " +
+ "memory replay can use when an event log is corrupt or unexpectedly large. Setting " +
+ "this to 0 or a negative value disables the limit. " +
+ "Introduced in 4.3.0; also available in 3.5.10, 4.0.5, 4.1.4 and 4.2.1; and in " +
+ "all versions after 4.3.0.")
+ .version("4.3.0")
+ .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
+ .bytesConf(ByteUnit.BYTE)
+ .createWithDefaultString("512m")
+
private[spark] val EVENT_LOG_ROLLING_MAX_FILES_TO_RETAIN =
ConfigBuilder("spark.history.fs.eventLog.rolling.maxFilesToRetain")
.doc("The maximum number of event log files which will be retained as non-compacted. " +
diff --git a/core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala b/core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala
index 2e6cfa98ff37..1cd00facb739 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala
@@ -17,22 +17,32 @@
package org.apache.spark.scheduler
-import java.io.{EOFException, InputStream, IOException}
+import java.io.{BufferedReader, EOFException, InputStream, InputStreamReader, IOException}
+import java.nio.charset.{CodingErrorAction, StandardCharsets}
-import scala.io.{Codec, Source}
+import scala.annotation.tailrec
import com.fasterxml.jackson.core.JsonParseException
import com.fasterxml.jackson.databind.exc.UnrecognizedPropertyException
+import org.apache.spark.SparkConf
import org.apache.spark.internal.Logging
import org.apache.spark.internal.LogKeys._
+import org.apache.spark.internal.config.History
import org.apache.spark.scheduler.ReplayListenerBus._
import org.apache.spark.util.JsonProtocol
/**
* A SparkListenerBus that can be used to replay events from serialized event data.
+ *
+ * @param maxLineLength Maximum number of byes (1/2 chars) of a single event log line that will be
+ * materialized during replay. Longer lines are drained, skipped and
+ * logged, bounding the memory replay can use when an event log is
+ * corrupt or unexpectedly large.
*/
-private[spark] class ReplayListenerBus extends SparkListenerBus with Logging {
+private[spark] class ReplayListenerBus(
+ maxLineLength: Int = ReplayListenerBus.DEFAULT_MAX_LINE_LENGTH)
+ extends SparkListenerBus with Logging {
/**
* Replay each event in the order maintained in the given stream. The stream is expected to
@@ -56,10 +66,79 @@ private[spark] class ReplayListenerBus extends SparkListenerBus with Logging {
sourceName: String,
maybeTruncated: Boolean = false,
eventsFilter: ReplayEventsFilter = SELECT_ALL_FILTER): Boolean = {
- val lines = Source.fromInputStream(logData)(Codec.UTF8).getLines()
+ val lines = boundedLines(logData, sourceName)
replay(lines, sourceName, maybeTruncated, eventsFilter)
}
+ /**
+ * Reads '\n'-terminated lines like Source.getLines(), but never materializes more than
+ * `maxLineLength` bytes of a single line. An over-long line is drained and skipped
+ * with a warning instead of being turned into a String.
+ */
+ private def boundedLines(logData: InputStream, sourceName: String): Iterator[String] = {
+ // Fail on malformed input like Source.getLines() does instead of replacing it.
+ val decoder = StandardCharsets.UTF_8.newDecoder()
+ .onMalformedInput(CodingErrorAction.REPORT)
+ .onUnmappableCharacter(CodingErrorAction.REPORT)
+ val reader = new BufferedReader(new InputStreamReader(logData, decoder))
+ new Iterator[String] {
+ private var nextLine: String = _
+ private var lineFetched = false
+ private var warned = false
+
+ override def hasNext: Boolean = {
+ if (!lineFetched) {
+ nextLine = fetchLine()
+ lineFetched = true
+ }
+ nextLine != null
+ }
+
+ override def next(): String = {
+ if (!hasNext) {
+ throw new NoSuchElementException("No more lines")
+ }
+ val line = nextLine
+ nextLine = null
+ lineFetched = false
+ line
+ }
+
+ @tailrec private def fetchLine(): String = {
+ val sb = new java.lang.StringBuilder()
+ var overLong = false
+ var c = reader.read()
+ if (c == -1) {
+ null
+ } else {
+ while (c != -1 && c != '\n') {
+ if (sb.length() * 2 < maxLineLength) {
+ sb.append(c.toChar)
+ } else {
+ overLong = true
+ }
+ c = reader.read()
+ }
+ if (overLong) {
+ if (!warned) {
+ logWarning(log"Skipped event log lines longer than " +
+ log"${MDC(MAX_SIZE, maxLineLength)} characters in " +
+ log"${MDC(FILE_NAME, sourceName)}")
+ warned = true
+ }
+ fetchLine()
+ } else {
+ // Handle CRLF line endings like Source.getLines() does.
+ if (sb.length() > 0 && sb.charAt(sb.length() - 1) == '\r') {
+ sb.setLength(sb.length() - 1)
+ }
+ sb.toString
+ }
+ }
+ }
+ }
+ }
+
/**
* Overloaded variant of [[replay()]] which accepts an iterator of lines instead of an
* [[InputStream]]. Exposed for use by custom ApplicationHistoryProvider implementations.
@@ -146,6 +225,19 @@ private[spark] class HaltReplayException extends RuntimeException
private[spark] object ReplayListenerBus {
+ /**
+ * Default per-line cap during replay: far above any legitimate event line, so replay
+ * memory stays bounded even for corrupt logs. Matches the default of
+ * spark.history.fs.eventLog.maxLineLength.
+ */
+ val DEFAULT_MAX_LINE_LENGTH: Int = 512 * 1024 * 1024
+
+ /** Resolves the replay line-length cap from configuration; <= 0 disables the cap. */
+ def maxLineLength(conf: SparkConf): Int = {
+ val configured = conf.get(History.EVENT_LOG_MAX_LINE_LENGTH)
+ if (configured <= 0 || configured > Int.MaxValue) Int.MaxValue else configured.toInt
+ }
+
type ReplayEventsFilter = (String) => Boolean
// utility filter that selects all event logs during replay
diff --git a/core/src/test/scala/org/apache/spark/scheduler/ReplayListenerSuite.scala b/core/src/test/scala/org/apache/spark/scheduler/ReplayListenerSuite.scala
index 1f71258bf867..f4652c23f63e 100644
--- a/core/src/test/scala/org/apache/spark/scheduler/ReplayListenerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/scheduler/ReplayListenerSuite.scala
@@ -78,6 +78,37 @@ class ReplayListenerSuite extends SparkFunSuite with BeforeAndAfter with LocalSp
assert(eventMonster.loggedEvents(1) === JsonProtocol.sparkEventToJsonString(applicationEnd))
}
+ test("Over-long event log lines are skipped instead of materialized") {
+ val logFilePath = getFilePath(testDir, "events.txt")
+ val fstream = fileSystem.create(logFilePath)
+ val fwriter = new OutputStreamWriter(fstream, StandardCharsets.UTF_8)
+ val applicationStart = SparkListenerApplicationStart("Greatest App (N)ever", None,
+ 125L, "Mickey", None)
+ val applicationEnd = SparkListenerApplicationEnd(1000L)
+ val maxLineLength = 8 * 1024
+ val hugeLine = "x" * (maxLineLength + 1024)
+ Utils.tryWithResource(new PrintWriter(fwriter)) { writer =>
+ // scalastyle:off println
+ writer.println(JsonProtocol.sparkEventToJsonString(applicationStart))
+ writer.println(hugeLine)
+ writer.println(JsonProtocol.sparkEventToJsonString(applicationEnd))
+ // scalastyle:on println
+ }
+
+ val logData = fileSystem.open(logFilePath)
+ val eventMonster = new EventBufferingListener
+ try {
+ val replayer = new ReplayListenerBus(maxLineLength = maxLineLength)
+ replayer.addListener(eventMonster)
+ assert(replayer.replay(logData, logFilePath.toString))
+ } finally {
+ logData.close()
+ }
+ assert(eventMonster.loggedEvents.size === 2)
+ assert(eventMonster.loggedEvents(0) === JsonProtocol.sparkEventToJsonString(applicationStart))
+ assert(eventMonster.loggedEvents(1) === JsonProtocol.sparkEventToJsonString(applicationEnd))
+ }
+
/**
* Test replaying compressed spark history file that internally throws an EOFException. To
* avoid sensitivity to the compression specifics the test forces an EOFException to occur
diff --git a/docs/monitoring.md b/docs/monitoring.md
index daf9cc2bfef4..83fd85065f44 100644
--- a/docs/monitoring.md
+++ b/docs/monitoring.md
@@ -438,6 +438,19 @@ Security options for the Spark History Server are covered more detail in the
</td>
<td>3.0.0</td>
</tr>
+ <tr>
+ <td>spark.history.fs.eventLog.maxLineLength</td>
+ <td>512m</td>
+ <td>
+ Maximum length of a single event log line during replay. Lines longer than this are
+ skipped with a warning instead of being read into memory, which bounds the memory replay
+ can use when an event log is corrupt or unexpectedly large. Setting this to 0 or a
+ negative value disables the limit.<br/>
+ Introduced in 4.3.0; also available in 3.5.10, 4.0.5, 4.1.4 and 4.2.1; and in all
+ versions after 4.3.0.
+ </td>
+ <td>4.3.0</td>
+ </tr>
<tr>
<td>spark.history.fs.eventLog.rolling.maxFilesToRetain</td>
<td>Int.MaxValue</td>
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org