viirya commented on code in PR #59328:
URL: https://github.com/apache/spark/pull/59328#discussion_r4235884263
##########
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:
Good catch, done. `initializeBuilders` now uses `forall` and stops at the
first file where replay returns `false` or skips a line, with a separate
warning for each case, and the result code is renamed to `INCOMPLETE_REPLAY`
since it is no longer only about over-long lines. Added a test with
`{"Event":"SparkListenerJobStart"}` (valid JSON, but `JsonProtocol` fails on
the missing `Job ID`) before a live job's start in the same file.
##########
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:
Agreed. I'll backport it 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`). I
also updated the PR description: the over-long line case is not in any released
version, but it is in the `v4.0.5-rc1`, `v4.1.4-rc2` and `v4.2.1-rc1` tags.
##########
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:
Yes, it lasts until the limit is increased. Compaction is evaluated once per
new rolled file, so it resumes with the next rolled file after that. I added
this to the PR description, the "Applying compaction on rolling event log
files" section, and the `maxLineLength` doc in both `docs/monitoring.md` and
`History.scala`, and the warning now says so too.
I'd rather not compact the files before the first file with a skipped line.
After that compaction, the range is the compact file plus the file with the
skipped line, so it is stuck again at the same place and the files after it
still grow. It only helps once and adds another code path.
##########
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:
Stopping inside the file would need a way to abort replay from the reader in
`ReplayListenerBus`, which I'd like to keep out of this fix since it is going
to all the maintenance branches. The extra read is at most the rest of one
rolled file (bounded by `spark.eventLog.rolling.maxFileSize`), while the files
before it are replayed on every attempt anyway. I updated the comment to say
that the stop happens between files.
##########
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:
Done, added a paragraph to "Applying compaction on rolling event log files"
and a sentence to the `spark.history.fs.eventLog.maxLineLength` entry (and the
config doc in `History.scala`).
##########
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:
Added:
- a test where the file after the one with the skipped line has a malformed
line, so replay would throw if it did not stop after the first file (it fails
if the per-file stop is removed),
- a test for replay that stops in the middle of a file (from your first
comment),
- a test in `SQLEventFilterBuilderSuite` for a live SQL execution with no
jobs whose start line exceeds the limit: compaction is skipped, and with the
default limit the start is kept in the compact file.
##########
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:
Done, the scaladoc now says that the `Iterator[String]` overload does not
apply the limit, so it never updates the count.
--
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]
