You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by vi...@apache.org on 2026/10/14 05:54:36 UTC
(spark) branch master updated: [SPARK-60110][UI] Skip event log compaction when replay skips lines over the line-length limit
This is an automated email from the ASF dual-hosted git repository.
viirya 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 bb462a1700ea [SPARK-60110][UI] Skip event log compaction when replay skips lines over the line-length limit
bb462a1700ea is described below
commit bb462a1700eaee72d26e9d9b507cf7bc8559c6f0
Author: Liang-Chi Hsieh <vi...@gmail.com>
AuthorDate: Tue Oct 13 22:54:23 2026 -0700
[SPARK-60110][UI] Skip event log compaction when replay skips lines over the line-length limit
### What changes were proposed in this pull request?
Event log compaction reads the files to compact twice: `EventLogFileCompactor.initializeBuilders` replays them through `ReplayListenerBus` to build the event filters, and `rewrite` applies the filters line by line. Since SPARK-59407, replay skips lines longer than `spark.history.fs.eventLog.maxLineLength`, but the filter builders are not aware of it.
This PR makes `EventLogFileCompactor.compact` skip compaction, keeping the original files, when replay does not fully read the files to compact:
- `ReplayListenerBus` counts the over-long lines it skips (`numSkippedLines`).
- `initializeBuilders` stops at the first file where replay skips a line or stops early on an error (`replay` returns `false`), logs a warning, and returns `None`. The check happens between files.
- `compact` then returns the new result code `CompactionResultCode.INCOMPLETE_REPLAY` without rewriting or deleting any file.
- `docs/monitoring.md` and the `spark.history.fs.eventLog.maxLineLength` doc describe the new skip condition.
When compaction proceeds, every line of the files to compact was read within the limit, so the reads in `rewrite` stay within it as well.
### Why are the changes needed?
The filter builders decide which jobs, stages, tasks and SQL executions are still live from the events they see. If a skipped line is the `SparkListenerJobStart` of a live job (for example, one with large job properties), the builders never learn that the job is live, so the rewrite drops the job start, leaving its stage and task events without the job. The same applies to a live `SparkListenerSQLExecutionStart` with a large plan. Compaction then deletes the original files, so these e [...]
The same loss happens, already before SPARK-59407, when replay stops in the middle of a file because a line is valid JSON but not a valid event (`replay` returns `false`), so this PR covers that case too.
Other options were considered:
- Applying the same limit in `rewrite`: both phases agree, but the dropped events are still lost permanently.
- Copying over-long lines to the compacted file verbatim: an over-long start of a finished job would be kept without its end event, so the job would appear running forever to a reader with a higher limit.
Skipping compaction is the only option that does not lose data. The cost is that, as the files to compact always start from the oldest one, compaction of such an application stays skipped until the limit is increased: the rolled files are not reduced to `spark.history.fs.eventLog.rolling.maxFilesToRetain`. Compaction is still evaluated once per new event log file, as for `LOW_SCORE_FOR_COMPACTION`, so it resumes with the next rolled file after the limit is increased. Compacting only t [...]
### Does this PR introduce _any_ user-facing change?
Yes. When event log compaction is enabled (`spark.history.fs.eventLog.rolling.maxFilesToRetain`) and the History Server cannot fully replay the files to compact, it no longer compacts the event log files of that application (it logs a warning instead), rather than compacting them and losing the events of live jobs and SQL executions. With an event log line over `spark.history.fs.eventLog.maxLineLength`, this lasts until the limit is increased.
No released version has the over-long line case, but SPARK-59407 is in the `v4.0.5-rc1`, `v4.1.4-rc1`/`v4.1.4-rc2` and `v4.2.1-rc1` tags, so this fix should be backported to the same branches as SPARK-59804 (`branch-4.x`, `branch-4.3`, `branch-4.2`, `branch-4.1`, `branch-4.0` and `branch-3.5`). The incomplete replay case exists in released versions.
### How was this patch tested?
New tests:
- `EventLogFileCompactorSuite`: a live job whose job start line exceeds the limit (compaction is skipped and all original files are kept; with the default limit the same files are compacted and the live job's events are kept), replay stops after the file with a skipped line (a later file in the range would make replay throw), and replay that stops in the middle of a file.
- `SQLEventFilterBuilderSuite`: a live SQL execution with no jobs whose start line exceeds the limit.
- `ReplayListenerSuite`: an assertion on `numSkippedLines`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 5.5)
Closes #59328 from viirya/SPARK-60110.
Authored-by: Liang-Chi Hsieh <vi...@gmail.com>
Signed-off-by: Liang-Chi Hsieh <vi...@gmail.com>
---
.../deploy/history/EventLogFileCompactor.scala | 66 ++++++++++---
.../org/apache/spark/internal/config/History.scala | 5 +-
.../apache/spark/scheduler/ReplayListenerBus.scala | 9 ++
.../history/EventLogFileCompactorSuite.scala | 107 ++++++++++++++++++++-
.../spark/deploy/history/EventLogTestHelper.scala | 20 +++-
.../spark/scheduler/ReplayListenerSuite.scala | 3 +-
docs/monitoring.md | 11 ++-
.../history/SQLEventFilterBuilderSuite.scala | 58 ++++++++++-
8 files changed, 254 insertions(+), 25 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 8d9be3877c78..2e54ae4db9b3 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
@@ -30,6 +30,7 @@ import org.apache.spark.SparkConf
import org.apache.spark.deploy.history.EventFilter.FilterStatistics
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.Utils
@@ -68,6 +69,8 @@ class EventLogFileCompactor(
* This method skips compaction for some circumstances described below:
* - not enough files on the range of compaction
* - score is lower than the threshold of compaction (meaning compaction won't help much)
+ * - replay does not fully read the files to compact (e.g. some lines exceed the event log
+ * line-length limit), so the filters cannot know whether the events are still needed
*
* If this method returns the compaction result as SUCCESS, caller needs to re-read the list
* of event log files, as new compact file is available as well as old event log files are
@@ -84,18 +87,24 @@ class EventLogFileCompactor(
if (filesToCompact.isEmpty) {
CompactionResult(CompactionResultCode.NOT_ENOUGH_FILES, None)
} else {
- val builders = initializeBuilders(fs, filesToCompact.map(_.getPath))
+ initializeBuilders(fs, filesToCompact.map(_.getPath)) match {
+ case None =>
+ CompactionResult(CompactionResultCode.INCOMPLETE_REPLAY, None)
- val filters = builders.map(_.createFilter())
- val minScore = filters.flatMap(_.statistics()).map(calculateScore).min
+ case Some(builders) =>
+ val filters = builders.map(_.createFilter())
+ val minScore = filters.flatMap(_.statistics()).map(calculateScore).min
- if (minScore < compactionThresholdScore) {
- CompactionResult(CompactionResultCode.LOW_SCORE_FOR_COMPACTION, None)
- } else {
- rewrite(filters, filesToCompact)
- cleanupCompactedFiles(filesToCompact)
- CompactionResult(CompactionResultCode.SUCCESS, Some(
- RollingEventLogFilesWriter.getEventLogFileIndex(filesToCompact.last.getPath.getName)))
+ if (minScore < compactionThresholdScore) {
+ CompactionResult(CompactionResultCode.LOW_SCORE_FOR_COMPACTION, None)
+ } else {
+ // Replay read every line of these files within the line-length limit, so the
+ // reads in rewrite stay within it as well.
+ rewrite(filters, filesToCompact)
+ cleanupCompactedFiles(filesToCompact)
+ CompactionResult(CompactionResultCode.SUCCESS, Some(
+ RollingEventLogFilesWriter.getEventLogFileIndex(filesToCompact.last.getPath.getName)))
+ }
}
}
}
@@ -111,21 +120,48 @@ class EventLogFileCompactor(
/**
* Loads all available EventFilterBuilders in classloader via ServiceLoader, and initializes
* them via replaying events in given files.
+ *
+ * Returns None if replay does not fully read the given files: a line over the line-length
+ * limit is skipped, or replay stops in the middle of a file on an error. The builders have not
+ * seen the missed events (e.g. the start of a live job), so filters created from them could
+ * reject events that are still needed, and those events would be lost once the original files
+ * are removed.
*/
- private def initializeBuilders(fs: FileSystem, files: Seq[Path]): Seq[EventFilterBuilder] = {
+ private def initializeBuilders(
+ fs: FileSystem,
+ files: Seq[Path]): Option[Seq[EventFilterBuilder]] = {
val bus = new ReplayListenerBus(ReplayListenerBus.maxLineLength(sparkConf))
val builders = ServiceLoader.load(classOf[EventFilterBuilder],
Utils.getContextOrSparkClassLoader).asScala.toSeq
builders.foreach(bus.addListener)
- files.foreach { log =>
- Utils.tryWithResource(EventLogFileReader.openEventLog(log, fs)) { in =>
+ // Stop at the first file that is not fully replayed, as compaction will not proceed. The
+ // check happens between files, so that file itself is still read to the end.
+ val replayedAll = files.forall { log =>
+ val completed = Utils.tryWithResource(EventLogFileReader.openEventLog(log, fs)) { in =>
bus.replay(in, log.getName)
}
+ if (bus.numSkippedLines > 0) {
+ logWarning(log"Skipping compaction of event log files in " +
+ log"${MDC(LogKeys.PATH, log.getParent)}: ${MDC(LogKeys.FILE_NAME, log.getName)} has " +
+ log"lines longer than ${MDC(LogKeys.CONFIG, History.EVENT_LOG_MAX_LINE_LENGTH.key)} " +
+ log"(${MDC(LogKeys.MAX_SIZE, ReplayListenerBus.maxLineLength(sparkConf))} bytes). " +
+ log"Compaction of this application stays skipped until the limit is increased and " +
+ log"a new event log file is rolled.")
+ false
+ } else if (!completed) {
+ logWarning(log"Skipping compaction of event log files in " +
+ log"${MDC(LogKeys.PATH, log.getParent)}: failed to replay " +
+ log"${MDC(LogKeys.FILE_NAME, log.getName)} entirely. Compaction of this application " +
+ log"stays skipped.")
+ false
+ } else {
+ true
+ }
}
- builders
+ if (replayedAll) Some(builders) else None
}
private def calculateScore(stats: FilterStatistics): Double = {
@@ -209,7 +245,7 @@ class EventLogFileCompactor(
case class CompactionResult(code: CompactionResultCode.Value, compactIndex: Option[Long])
object CompactionResultCode extends Enumeration {
- val SUCCESS, NOT_ENOUGH_FILES, LOW_SCORE_FOR_COMPACTION = Value
+ val SUCCESS, NOT_ENOUGH_FILES, LOW_SCORE_FOR_COMPACTION, INCOMPLETE_REPLAY = Value
}
/**
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 b75500f7e5c3..53f214d916b3 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
@@ -177,7 +177,10 @@ private[spark] object History {
"the retained buffer; use 128m to reach a smaller capacity. Values at or below 0, " +
s"or above $EVENT_LOG_MAX_LINE_LENGTH_LIMIT, use the maximum supported limit of " +
s"$EVENT_LOG_MAX_LINE_LENGTH_LIMIT bytes (512 MiB). This cap avoids JVM array-size " +
- "limits but does not guarantee sufficient heap space. " +
+ "limits but does not guarantee sufficient heap space. If a longer line is in the " +
+ "event log files to compact, compaction of the rolling event log files of that " +
+ "application is skipped until the limit is increased and a new event log file is " +
+ "rolled. " +
"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")
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 24b7e2a9d3d6..492a39ea5a2b 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala
@@ -50,6 +50,14 @@ private[spark] class ReplayListenerBus(
private[scheduler] val effectiveMaxLineLength =
ReplayListenerBus.normalizeMaxLineLength(maxLineLength)
+ private var skippedLines = 0L
+
+ /**
+ * Number of over-long lines skipped so far by replay calls on an [[InputStream]]. The
+ * `Iterator[String]` overload does not apply the limit, so it never updates this count.
+ */
+ private[spark] def numSkippedLines: Long = skippedLines
+
/**
* Replay each event in the order maintained in the given stream. The stream is expected to
* contain one JSON-encoded SparkListenerEvent per line.
@@ -166,6 +174,7 @@ private[spark] class ReplayListenerBus(
log"${MDC(LINE_NUM, index + 1)}" + lineOrigin(startOffset))
warned = true
}
+ skippedLines += 1
logDebug(s"Skipped event log line ${index + 1} in $sourceName" +
lineOrigin(startOffset).message)
fetchLine()
diff --git a/core/src/test/scala/org/apache/spark/deploy/history/EventLogFileCompactorSuite.scala b/core/src/test/scala/org/apache/spark/deploy/history/EventLogFileCompactorSuite.scala
index 5297ac5aac89..c56fe40dcca8 100644
--- a/core/src/test/scala/org/apache/spark/deploy/history/EventLogFileCompactorSuite.scala
+++ b/core/src/test/scala/org/apache/spark/deploy/history/EventLogFileCompactorSuite.scala
@@ -17,6 +17,8 @@
package org.apache.spark.deploy.history
+import java.util.Properties
+
import scala.collection.mutable
import scala.io.{Codec, Source}
@@ -24,7 +26,8 @@ import org.apache.hadoop.fs.{FileStatus, FileSystem, Path}
import org.apache.spark.{SparkConf, SparkFunSuite}
import org.apache.spark.deploy.SparkHadoopUtil
-import org.apache.spark.deploy.history.EventLogTestHelper.writeEventsToRollingWriter
+import org.apache.spark.deploy.history.EventLogTestHelper.{convertEvent, writeEventsToRollingWriter, writeLinesToRollingWriter}
+import org.apache.spark.internal.config.History
import org.apache.spark.scheduler._
import org.apache.spark.scheduler.cluster.ExecutorInfo
import org.apache.spark.status.ListenerEventsTestHelper._
@@ -202,6 +205,108 @@ class EventLogFileCompactorSuite extends SparkFunSuite {
}
}
+ test("SPARK-60110: Don't compact files if some lines exceed the line-length limit") {
+ withTempDir { dir =>
+ val fs = new Path(dir.getAbsolutePath).getFileSystem(hadoopConf)
+
+ // job 1 is finished
+ val stage1 = createStage(1, createRddsWithId(1 to 2), Nil)
+ val tasks1 = createTasks(4, Array("exec1"), 0L).map(createTaskStartEvent(_, 1, 0))
+
+ // job 2 is still live, and its job start line exceeds the line-length limit
+ val stage2 = createStage(2, createRddsWithId(3 to 4), Nil)
+ val tasks2 = createTasks(4, Array("exec1"), 4L).map(createTaskStartEvent(_, 2, 0))
+ val maxLineLength = 8 * 1024
+ val props = new Properties()
+ props.setProperty("large", "x" * maxLineLength)
+ val jobStart2 = SparkListenerJobStart(2, 0, Seq(stage2), props)
+ val stageSubmitted2 = SparkListenerStageSubmitted(stage2)
+
+ // 1~3 are candidates to compact, 4~6 are dummies to ensure max files to retain
+ val fileStatuses = writeEventsToRollingWriter(fs, "app", dir, sparkConf, hadoopConf,
+ Seq(SparkListenerJobStart(1, 0, Seq(stage1)), SparkListenerStageSubmitted(stage1)) ++
+ tasks1,
+ Seq(jobStart2, stageSubmitted2) ++ tasks2,
+ Seq(SparkListenerJobEnd(1, 0, JobSucceeded)),
+ testEvent,
+ testEvent,
+ testEvent)
+
+ // Replay skips job 2's start, so the filters would not know that job 2 is live and the
+ // rewrite would drop its job start, leaving its stage and task events without the job.
+ // Compaction must not proceed.
+ val limitedConf = sparkConf.clone()
+ .set(History.EVENT_LOG_MAX_LINE_LENGTH, maxLineLength.toLong)
+ val limitedCompactor = new EventLogFileCompactor(limitedConf, hadoopConf, fs,
+ TEST_ROLLING_MAX_FILES_TO_RETAIN, TEST_COMPACTION_SCORE_THRESHOLD)
+ assertNoCompaction(fs, fileStatuses, limitedCompactor.compact(fileStatuses),
+ CompactionResultCode.INCOMPLETE_REPLAY)
+
+ // With the default limit, the events of live job 2 are kept in the compact file.
+ val compactor = new EventLogFileCompactor(sparkConf, hadoopConf, fs,
+ TEST_ROLLING_MAX_FILES_TO_RETAIN, TEST_COMPACTION_SCORE_THRESHOLD)
+ assertCompaction(fs, fileStatuses, compactor.compact(fileStatuses),
+ expectedNumOfFilesCompacted = 3)
+
+ val compactFilePath = getCompactFilePath(fileStatuses(2).getPath)
+ Utils.tryWithResource(EventLogFileReader.openEventLog(compactFilePath, fs)) { is =>
+ val lines = Source.fromInputStream(is)(Codec.UTF8).getLines().toList
+ val expectedLines = (Seq(jobStart2, stageSubmitted2) ++ tasks2).map(convertEvent)
+ assert(lines === expectedLines)
+ }
+ }
+ }
+
+ test("SPARK-60110: Stop replaying files to compact after the file with a skipped line") {
+ withTempDir { dir =>
+ val fs = new Path(dir.getAbsolutePath).getFileSystem(hadoopConf)
+ val maxLineLength = 8 * 1024
+ val testLine = convertEvent(testEvent.head)
+
+ // 1~2 are candidates to compact. Replay would throw on the malformed line in file 2 if it
+ // were not stopped after file 1.
+ val fileStatuses = writeLinesToRollingWriter(fs, "app", dir, sparkConf, hadoopConf,
+ Seq(testLine, "x" * (maxLineLength + 1)),
+ Seq("{"),
+ Seq(testLine),
+ Seq(testLine),
+ Seq(testLine))
+
+ val conf = sparkConf.clone().set(History.EVENT_LOG_MAX_LINE_LENGTH, maxLineLength.toLong)
+ val compactor = new EventLogFileCompactor(conf, hadoopConf, fs,
+ TEST_ROLLING_MAX_FILES_TO_RETAIN, TEST_COMPACTION_SCORE_THRESHOLD)
+ assertNoCompaction(fs, fileStatuses, compactor.compact(fileStatuses),
+ CompactionResultCode.INCOMPLETE_REPLAY)
+ }
+ }
+
+ test("SPARK-60110: Don't compact files if replay stops in the middle of a file") {
+ withTempDir { dir =>
+ val fs = new Path(dir.getAbsolutePath).getFileSystem(hadoopConf)
+
+ // job 1 is still live, but its job start comes after a line that is valid JSON but not a
+ // valid event, which makes replay stop reading the rest of file 1
+ val stage1 = createStage(1, createRddsWithId(1 to 2), Nil)
+ val tasks1 = createTasks(4, Array("exec1"), 0L).map(createTaskStartEvent(_, 1, 0))
+ val testLine = convertEvent(testEvent.head)
+
+ // 1~2 are candidates to compact, 3~5 are dummies to ensure max files to retain
+ val fileStatuses = writeLinesToRollingWriter(fs, "app", dir, sparkConf, hadoopConf,
+ Seq("{\"Event\":\"SparkListenerJobStart\"}") ++
+ (Seq(SparkListenerJobStart(1, 0, Seq(stage1)), SparkListenerStageSubmitted(stage1)) ++
+ tasks1).map(convertEvent),
+ Seq(testLine),
+ Seq(testLine),
+ Seq(testLine),
+ Seq(testLine))
+
+ val compactor = new EventLogFileCompactor(sparkConf, hadoopConf, fs,
+ TEST_ROLLING_MAX_FILES_TO_RETAIN, TEST_COMPACTION_SCORE_THRESHOLD)
+ assertNoCompaction(fs, fileStatuses, compactor.compact(fileStatuses),
+ CompactionResultCode.INCOMPLETE_REPLAY)
+ }
+ }
+
test("rewrite files with test filters") {
class TestEventFilter1 extends EventFilter {
override def acceptFn(): PartialFunction[SparkListenerEvent, Boolean] = {
diff --git a/core/src/test/scala/org/apache/spark/deploy/history/EventLogTestHelper.scala b/core/src/test/scala/org/apache/spark/deploy/history/EventLogTestHelper.scala
index 0161917f8853..26dd6e2f3155 100644
--- a/core/src/test/scala/org/apache/spark/deploy/history/EventLogTestHelper.scala
+++ b/core/src/test/scala/org/apache/spark/deploy/history/EventLogTestHelper.scala
@@ -85,14 +85,24 @@ object EventLogTestHelper {
sparkConf: SparkConf,
hadoopConf: Configuration,
eventsFiles: Seq[SparkListenerEvent]*): Seq[FileStatus] = {
+ writeLinesToRollingWriter(fs, appId, dir, sparkConf, hadoopConf,
+ eventsFiles.map(_.map(convertEvent)): _*)
+ }
+
+ /** Writes each sequence of lines to its own rolled event log file, as they are. */
+ def writeLinesToRollingWriter(
+ fs: FileSystem,
+ appId: String,
+ dir: File,
+ sparkConf: SparkConf,
+ hadoopConf: Configuration,
+ linesFiles: Seq[String]*): Seq[FileStatus] = {
val writer = new RollingEventLogFilesWriter(appId, None, dir.toURI, sparkConf, hadoopConf)
writer.start()
- eventsFiles.dropRight(1).foreach { events =>
- writeEventsToRollingWriter(writer, events, rollFile = true)
- }
- eventsFiles.lastOption.foreach { events =>
- writeEventsToRollingWriter(writer, events, rollFile = false)
+ linesFiles.zipWithIndex.foreach { case (lines, index) =>
+ lines.foreach(writer.writeEvent(_, flushLogger = true))
+ if (index < linesFiles.length - 1) writer.rollEventLogFile()
}
writer.stop()
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 3b646a816b28..8042d1eb7574 100644
--- a/core/src/test/scala/org/apache/spark/scheduler/ReplayListenerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/scheduler/ReplayListenerSuite.scala
@@ -101,13 +101,14 @@ class ReplayListenerSuite extends SparkFunSuite with BeforeAndAfter with LocalSp
val logData = fileSystem.open(logFilePath)
val eventMonster = new EventBufferingListener
+ val replayer = new ReplayListenerBus(maxLineLength = maxLineLength)
try {
- val replayer = new ReplayListenerBus(maxLineLength = maxLineLength)
replayer.addListener(eventMonster)
assert(replayer.replay(logData, logFilePath.toString))
} finally {
logData.close()
}
+ assert(replayer.numSkippedLines === 1)
assert(eventMonster.loggedEvents.size === 2)
assert(eventMonster.loggedEvents(0) === JsonProtocol.sparkEventToJsonString(applicationStart))
assert(eventMonster.loggedEvents(1) === JsonProtocol.sparkEventToJsonString(applicationEnd))
diff --git a/docs/monitoring.md b/docs/monitoring.md
index 55cf5d778509..ded2e48ae2f7 100644
--- a/docs/monitoring.md
+++ b/docs/monitoring.md
@@ -128,6 +128,12 @@ The compaction tries to exclude the events which point to the outdated data. As
* Events for the executor which is terminated
* Events for the SQL execution which is finished, and related job/stage/tasks events
+Spark History Server skips the compaction if it cannot read all the events in the target files, since it
+would not be able to tell which events are still needed. As the target files always start from the oldest one,
+compaction of that application then stays skipped. If an event log line in the target files is longer than
+<code>spark.history.fs.eventLog.maxLineLength</code>, compaction is skipped until the limit is increased and
+a new event log file is rolled.
+
Once rewriting is done, original log files will be deleted, via best-effort manner. The History Server may not be able to delete
the original log files, but it will not affect the operation of the History Server.
@@ -449,7 +455,10 @@ Security options for the Spark History Server are covered more detail in the
in steps: reducing 256m to 200m or 150m may not reduce the retained buffer; use 128m to
reach a smaller capacity. Values at or below
0, or above 536870912, use the maximum supported limit of 536870912 bytes (512 MiB).
- This cap avoids JVM array-size limits but does not guarantee sufficient heap space.<br/>
+ This cap avoids JVM array-size limits but does not guarantee sufficient heap space.
+ If a longer line is in the event log files to compact, compaction of the rolling event
+ log files of that application is skipped until the limit is increased and a new event log
+ file is rolled.<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>
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/history/SQLEventFilterBuilderSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/history/SQLEventFilterBuilderSuite.scala
index 90164234c338..76600209d1a0 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/history/SQLEventFilterBuilderSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/history/SQLEventFilterBuilderSuite.scala
@@ -19,11 +19,19 @@ package org.apache.spark.sql.execution.history
import java.util.Properties
-import org.apache.spark.SparkFunSuite
+import scala.io.{Codec, Source}
+
+import org.apache.hadoop.fs.Path
+
+import org.apache.spark.{SparkConf, SparkFunSuite}
+import org.apache.spark.deploy.SparkHadoopUtil
+import org.apache.spark.deploy.history.{CompactionResultCode, EventLogFileCompactor, EventLogFileReader, EventLogFileWriter, EventLogTestHelper}
+import org.apache.spark.internal.config.History
import org.apache.spark.scheduler._
import org.apache.spark.sql.execution.{SparkPlanInfo, SQLExecution}
import org.apache.spark.sql.execution.ui.{SparkListenerSQLExecutionEnd, SparkListenerSQLExecutionStart}
import org.apache.spark.status.ListenerEventsTestHelper
+import org.apache.spark.util.Utils
class SQLEventFilterBuilderSuite extends SparkFunSuite {
import ListenerEventsTestHelper._
@@ -105,4 +113,52 @@ class SQLEventFilterBuilderSuite extends SparkFunSuite {
assert(listener.liveTasks.isEmpty)
assert(listener.liveRDDs.isEmpty)
}
+
+ test("SPARK-60110: Don't compact files if a live SQL execution start is skipped") {
+ withTempDir { dir =>
+ val sparkConf = new SparkConf()
+ val hadoopConf = SparkHadoopUtil.newConfiguration(sparkConf)
+ val fs = new Path(dir.getAbsolutePath).getFileSystem(hadoopConf)
+ val planInfo = new SparkPlanInfo("node", "str", Seq.empty, Map.empty, Seq.empty)
+ val maxLineLength = 8 * 1024
+
+ // SQL execution 1 is still live and has no jobs yet, and its start line exceeds the
+ // line-length limit. SQL execution 2 is finished.
+ val sqlStart1 = SparkListenerSQLExecutionStart(1, Some(1), "desc1", "details1",
+ "x" * maxLineLength, planInfo, 0L)
+ val sqlStart2 = SparkListenerSQLExecutionStart(2, Some(2), "desc2", "details2", "plan",
+ planInfo, 0L)
+ val appStart = SparkListenerApplicationStart("app", Some("app"), 0, "user", None)
+
+ // 1~2 are candidates to compact, 3~5 are dummies to ensure max files to retain
+ val fileStatuses = EventLogTestHelper.writeEventsToRollingWriter(fs, "app", dir,
+ sparkConf, hadoopConf,
+ Seq(sqlStart1),
+ Seq(sqlStart2, SparkListenerSQLExecutionEnd(2, 0L)),
+ Seq(appStart),
+ Seq(appStart),
+ Seq(appStart))
+ val maxFilesToRetain = 3
+
+ // Replay skips the start of SQL execution 1, so the filters would treat it as finished
+ // and drop it. Compaction must not proceed.
+ val limitedConf = sparkConf.clone()
+ .set(History.EVENT_LOG_MAX_LINE_LENGTH, maxLineLength.toLong)
+ val limitedCompactor = new EventLogFileCompactor(limitedConf, hadoopConf, fs,
+ maxFilesToRetain, 0.0d)
+ val result = limitedCompactor.compact(fileStatuses)
+ assert(result.code === CompactionResultCode.INCOMPLETE_REPLAY)
+ fileStatuses.foreach { status => assert(fs.exists(status.getPath)) }
+
+ // With the default limit, the start of live SQL execution 1 is kept in the compact file.
+ val compactor = new EventLogFileCompactor(sparkConf, hadoopConf, fs, maxFilesToRetain, 0.0d)
+ assert(compactor.compact(fileStatuses).code === CompactionResultCode.SUCCESS)
+ val compactFilePath = new Path(fileStatuses(1).getPath.getParent,
+ fileStatuses(1).getPath.getName + EventLogFileWriter.COMPACTED)
+ Utils.tryWithResource(EventLogFileReader.openEventLog(compactFilePath, fs)) { is =>
+ val lines = Source.fromInputStream(is)(Codec.UTF8).getLines().toList
+ assert(lines === Seq(EventLogTestHelper.convertEvent(sqlStart1)))
+ }
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org