dongjoon-hyun commented on code in PR #59328:
URL: https://github.com/apache/spark/pull/59328#discussion_r4235443272
##########
core/src/main/scala/org/apache/spark/deploy/history/EventLogFileCompactor.scala:
##########
@@ -111,21 +125,29 @@ class EventLogFileCompactor(
/**
* Loads all available EventFilterBuilders in classloader via ServiceLoader,
and initializes
* them via replaying events in given files.
+ *
+ * Returns None if replay skips any line over the line-length limit. The
builders have not seen
+ * the skipped 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 =>
+ // Stop replaying once a line is skipped, as compaction will not proceed.
+ files.iterator.takeWhile(_ => bus.numSkippedLines == 0).foreach { log =>
Utils.tryWithResource(EventLogFileReader.openEventLog(log, fs)) { in =>
bus.replay(in, log.getName)
Review Comment:
The return value of `bus.replay(...)` is ignored here. `replayEntries`
catches generic exceptions and returns `false` without replaying the rest of
the file. For example, a line that is valid JSON but fails in
`JsonProtocol.sparkEventFromJson` because a field is missing can cause this. If
a live job's `SparkListenerJobStart` comes after that line in the same file,
the builders never see it. `numSkippedLines` is still `0`, so compaction goes
ahead: `rewrite` rejects the live job's events, and the original files are
deleted. That is the same data loss this PR fixes for over-long lines.
This was already the case before this PR, but should the check cover any
incomplete replay instead of only skipped lines? For example:
```scala
val complete = files.forall { log =>
Utils.tryWithResource(EventLogFileReader.openEventLog(log, fs)) { in =>
bus.replay(in, log.getName)
} && bus.numSkippedLines == 0
}
if (complete) Some(builders) else None
```
This would also stop at the first incomplete file, and `forall` would
replace `takeWhile`.
##########
core/src/main/scala/org/apache/spark/deploy/history/EventLogFileCompactor.scala:
##########
@@ -84,18 +87,29 @@ class EventLogFileCompactor(
if (filesToCompact.isEmpty) {
CompactionResult(CompactionResultCode.NOT_ENOUGH_FILES, None)
} else {
- val builders = initializeBuilders(fs, filesToCompact.map(_.getPath))
-
- 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)))
+ initializeBuilders(fs, filesToCompact.map(_.getPath)) match {
+ case None =>
Review Comment:
Once an over-long line is in the range to compact, compaction of this
application never runs again. The range always starts from the first (or
compact) file, so the over-long line stays in it. For a long-running streaming
application (the main use case of rolling event logs and compaction), this
means:
- `spark.history.fs.eventLog.rolling.maxFilesToRetain` is silently not
enforced, and the rolled files grow without bound. The application UI also has
to replay all of them.
- Each new rolled file replays the same files again and logs two WARN
messages (one from `ReplayListenerBus` and this one).
- `FsHistoryProvider.compact` ignores the result code, so the listing does
not show that the application is in this state.
This might be an acceptable trade-off, but could we at least say in the
description and docs that it is permanent for the application? Another option
is to compact the files before the first file with a skipped line, so that the
files before it are still compacted.
##########
core/src/main/scala/org/apache/spark/deploy/history/EventLogFileCompactor.scala:
##########
@@ -84,18 +87,29 @@ class EventLogFileCompactor(
if (filesToCompact.isEmpty) {
CompactionResult(CompactionResultCode.NOT_ENOUGH_FILES, None)
} else {
- val builders = initializeBuilders(fs, filesToCompact.map(_.getPath))
-
- 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)))
+ initializeBuilders(fs, filesToCompact.map(_.getPath)) match {
+ case None =>
+ logWarning(log"Skipping compaction of event log files in " +
+ log"${MDC(LogKeys.PATH, filesToCompact.head.getPath.getParent)}:
some lines exceed " +
+ log"${MDC(LogKeys.CONFIG, History.EVENT_LOG_MAX_LINE_LENGTH.key)}
" +
+ log"(${MDC(LogKeys.MAX_SIZE,
ReplayListenerBus.maxLineLength(sparkConf))} bytes). " +
+ log"Increase the limit to compact these files.")
+ CompactionResult(CompactionResultCode.LINES_OVER_LENGTH_LIMIT, None)
Review Comment:
The PR description says that no released version is affected. However,
SPARK-59407 is already in the `v4.0.5-rc1`, `v4.1.4-rc2` and `v4.2.1-rc1` tags,
and the config doc also lists 3.5.10. If those RCs are released without this
fix, the data loss will be in released versions. Could you backport this to
`branch-4.2`, `branch-4.1`, `branch-4.0` and `branch-3.5` (and
`branch-4.x`/`branch-4.3`) before the RCs are released?
##########
core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala:
##########
@@ -50,6 +50,11 @@ 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]]. */
+ private[spark] def numSkippedLines: Long = skippedLines
Review Comment:
nit. `numSkippedLines` only counts lines skipped by `replay(InputStream,
...)`. The public `replay(Iterator[String], ...)` overload neither applies the
limit nor updates the counter. So a caller that replays through that overload
and checks `numSkippedLines == 0`, as the compactor does, cannot tell "nothing
skipped" from "not tracked". The scaladoc mentions this, but it is easy to
miss. Should the doc say clearly that the value is always `0` for the
`Iterator[String]` overload?
##########
core/src/main/scala/org/apache/spark/deploy/history/EventLogFileCompactor.scala:
##########
@@ -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)
+ * - some lines exceed the event log line-length limit, so replay skips them
and the filters
Review Comment:
This is a user-facing change, but `docs/monitoring.md` is not updated.
Neither the "Applying compaction on rolling event log files" section nor the
`spark.history.fs.eventLog.maxLineLength` entry mentions that a line over the
limit stops compaction of that application. Could we add a sentence to both?
Operators who lower `maxLineLength` to bound History Server memory would need
to know about this.
##########
core/src/main/scala/org/apache/spark/deploy/history/EventLogFileCompactor.scala:
##########
@@ -111,21 +125,29 @@ class EventLogFileCompactor(
/**
* Loads all available EventFilterBuilders in classloader via ServiceLoader,
and initializes
* them via replaying events in given files.
+ *
+ * Returns None if replay skips any line over the line-length limit. The
builders have not seen
+ * the skipped 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 =>
+ // Stop replaying once a line is skipped, as compaction will not proceed.
+ files.iterator.takeWhile(_ => bus.numSkippedLines == 0).foreach { log =>
Review Comment:
nit. `takeWhile` checks the counter only between files. The rest of the file
with the skipped line is still decompressed, parsed and posted to every
builder, and the result is then discarded. This is repeated on every new rolled
file (see the comment on L91). Could we stop inside the file as well, or at
least say in the comment that the stop happens per file?
##########
core/src/test/scala/org/apache/spark/deploy/history/EventLogFileCompactorSuite.scala:
##########
@@ -202,6 +205,57 @@ 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 treat job 2 as
finished and drop
+ // all of its events. 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),
Review Comment:
This checks the result code and that the files are unchanged. It doesn't
check that replay stops after the file with the skipped line. If the
`takeWhile` is removed, the result is still `LINES_OVER_LENGTH_LIMIT` and this
test still passes. The live `SparkListenerSQLExecutionStart` case from the PR
description is also not covered (for example in `sql/core`, next to the
existing SQL event filter tests). Could we add tests for these?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
