(spark) branch master updated: [SPARK-52458][CORE] Support `spark.eventLog.excludedPatterns`

[do...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

dongjoon 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 1d4d06b3dd31 [SPARK-52458][CORE] Support `spark.eventLog.excludedPatterns`
1d4d06b3dd31 is described below

commit 1d4d06b3dd3185aa88dde7bd933c5cb49b7735d3
Author: Dongjoon Hyun <do...@apache.org>
AuthorDate: Wed Jun 11 22:38:34 2025 -0700

    [SPARK-52458][CORE] Support `spark.eventLog.excludedPatterns`
    
    ### What changes were proposed in this pull request?
    
    This PR aims to support `spark.eventLog.excludedPatterns` to exclude specific `SparkEvent`s. This has two goals.
    1. Save the cost of event log processing.
    2. Provide a full control to save or not on top of the existing `logEvent`.
    
    https://github.com/apache/spark/blob/68136fd50a5b635ed5f1a17984b79d86bdc357eb/common/utils/src/main/scala/org/apache/spark/scheduler/SparkListenerEvent.scala#L25-L28
    
    ### Why are the changes needed?
    
    Historically, Apache Spark provides multiple ways to manage the event logs to save a storage cost.
    - `spark.history.fs.cleaner.maxAge`: Delete old Spark jobs by age
    - `spark.history.fs.cleaner.maxNum`: Delete old Spark jobs by the total number of jobs
    - `spark.history.fs.eventLog.rolling.maxFilesToRetain`: Decompress + Compact + Compress back
    
    For example, after compaction, Spark event logs only have the following.
    
    ```
    // The extracted event names from a `compacted` event log file
    "SparkListenerLogStart"
    "SparkListenerResourceProfileAdded"
    "org.apache.spark.sql.connect.service.SparkListenerConnectServiceStarted"
    "SparkListenerBlockManagerAdded"
    "SparkListenerEnvironmentUpdate"
    "SparkListenerApplicationStart"
    "SparkListenerExecutorAdded"
    "SparkListenerBlockManagerAdded"
    "SparkListenerExecutorAdded"
    "SparkListenerBlockManagerAdded"
    "SparkListenerJobStart"
    "SparkListenerStageSubmitted"
    "SparkListenerTaskStart"
    "SparkListenerTaskEnd"
    "SparkListenerStageCompleted"
    ```
    
    This PR aims to provide a simple alternative to allow the users to skip specific Spark events completely.
    
    ```
    org.apache.spark.sql.execution.ui.SparkListenerSQLAdaptiveExecutionUpdate
    org.apache.spark.sql.execution.ui.SparkListenerSQLExecutionStart
    org.apache.spark.sql.execution.ui.SparkListenerSQLExecutionEnd
    org.apache.spark.sql.execution.ui.SparkListenerDriverAccumUpdates
    ```
    
    ### Does this PR introduce _any_ user-facing change?
    
    No. This is a new feature.
    
    ### How was this patch tested?
    
    Pass the CIs with the newly added test case.
    
    ### Was this patch authored or co-authored using generative AI tooling?
    
    No.
    
    Closes #51163 from dongjoon-hyun/SPARK-52458.
    
    Authored-by: Dongjoon Hyun <do...@apache.org>
    Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
 .../spark/deploy/history/EventLogFileWriters.scala   |  4 ++++
 .../org/apache/spark/internal/config/package.scala   |  8 ++++++++
 .../deploy/history/EventLogFileWritersSuite.scala    | 20 ++++++++++++++++++++
 3 files changed, 32 insertions(+)

diff --git a/core/src/main/scala/org/apache/spark/deploy/history/EventLogFileWriters.scala b/core/src/main/scala/org/apache/spark/deploy/history/EventLogFileWriters.scala
index 990ab680f3aa..83bc1443112a 100644
--- a/core/src/main/scala/org/apache/spark/deploy/history/EventLogFileWriters.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/history/EventLogFileWriters.scala
@@ -43,6 +43,7 @@ import org.apache.spark.util.Utils
  *   spark.eventLog.compression.codec - The codec to compress logged events
  *   spark.eventLog.overwrite - Whether to overwrite any existing files
  *   spark.eventLog.buffer.kb - Buffer size to use when writing to output streams
+ *   spark.eventLog.excludedPatterns - Specifes a comma-separated event names to be excluded
  *
  * Note that descendant classes can maintain its own parameters: refer the javadoc of each class
  * for more details.
@@ -58,6 +59,8 @@ abstract class EventLogFileWriter(
 
   protected val shouldCompress = sparkConf.get(EVENT_LOG_COMPRESS) &&
       !sparkConf.get(EVENT_LOG_COMPRESSION_CODEC).equalsIgnoreCase("none")
+  protected val excludedPatterns = sparkConf.get(EVENT_LOG_EXCLUDED_PATTERNS)
+      .map(name => s"""{"Event":"$name"""")
   protected val shouldOverwrite = sparkConf.get(EVENT_LOG_OVERWRITE)
   protected val outputBufferSize = sparkConf.get(EVENT_LOG_OUTPUT_BUFFER_SIZE).toInt * 1024
   protected val fileSystem = Utils.getHadoopFileSystem(logBaseDir, hadoopConf)
@@ -117,6 +120,7 @@ abstract class EventLogFileWriter(
   }
 
   protected def writeLine(line: String, flushLogger: Boolean = false): Unit = {
+    if (excludedPatterns.exists(line.startsWith(_))) return
     // scalastyle:off println
     writer.foreach(_.println(line))
     // scalastyle:on println
diff --git a/core/src/main/scala/org/apache/spark/internal/config/package.scala b/core/src/main/scala/org/apache/spark/internal/config/package.scala
index 000de4d6c26d..db9bb89be4ed 100644
--- a/core/src/main/scala/org/apache/spark/internal/config/package.scala
+++ b/core/src/main/scala/org/apache/spark/internal/config/package.scala
@@ -224,6 +224,14 @@ package object config {
       .booleanConf
       .createWithDefault(false)
 
+  private[spark] val EVENT_LOG_EXCLUDED_PATTERNS =
+    ConfigBuilder("spark.eventLog.excludedPatterns")
+      .doc("Specifies comma-separated event names to be excluded from the event logs.")
+      .version("4.1.0")
+      .stringConf
+      .toSequence
+      .createWithDefault(Nil)
+
   private[spark] val EVENT_LOG_ALLOW_EC =
     ConfigBuilder("spark.eventLog.erasureCoding.enabled")
       .version("3.0.0")
diff --git a/core/src/test/scala/org/apache/spark/deploy/history/EventLogFileWritersSuite.scala b/core/src/test/scala/org/apache/spark/deploy/history/EventLogFileWritersSuite.scala
index da3614bf81a5..a2718b973b15 100644
--- a/core/src/test/scala/org/apache/spark/deploy/history/EventLogFileWritersSuite.scala
+++ b/core/src/test/scala/org/apache/spark/deploy/history/EventLogFileWritersSuite.scala
@@ -112,6 +112,26 @@ abstract class EventLogFileWritersSuite extends SparkFunSuite with LocalSparkCon
     assert(writer.compressionCodecName === EVENT_LOG_COMPRESSION_CODEC.defaultValue)
   }
 
+  test("SPARK-52458: Support spark.eventLog.excludedPatterns") {
+    val appId = getUniqueApplicationId
+    val attemptId = None
+
+    val conf = getLoggingConf(testDirPath, None)
+    conf.set(EVENT_LOG_EXCLUDED_PATTERNS, Seq("B", "C"))
+
+    val writer = createWriter(appId, attemptId, testDirPath.toUri, conf,
+      SparkHadoopUtil.get.newConfiguration(conf))
+
+    writer.start()
+    Seq("A", "B", "C", "D").foreach { name =>
+      writer.writeEvent(s"""{"Event":"$name"}""", flushLogger = true)
+    }
+    writer.stop()
+
+    verifyWriteEventLogFile(appId, attemptId, testDirPath.toUri,
+      None, Seq("""{"Event":"A"}""", """{"Event":"D"}"""))
+  }
+
   protected def readLinesFromEventLogFile(log: Path, fs: FileSystem): List[String] = {
     val logDataStream = EventLogFileReader.openEventLog(log, fs)
     try {


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org