Fix capturing the history event for the initial config setup.

Netflix-Skunkworks · May 1, 2024 · ee83718 · ee83718
1 parent 5f18228
commit ee83718
Show file tree

Hide file tree

Showing 4 changed files with 12 additions and 5 deletions.
diff --git a/src/main/scala/com/qubole/sparklens/QuboleJobListener.scala b/src/main/scala/com/qubole/sparklens/QuboleJobListener.scala
@@ -131,8 +131,10 @@ class QuboleJobListener(sparkConf: SparkConf)  extends SparkListener {
     }
   }
 
+  override def onOtherEvent(event: SparkListenerEvent): Unit = {
+  }
+
   override def onEnvironmentUpdate(environmentUpdate: SparkListenerEnvironmentUpdate): Unit = {
-    println("Yay env update.")
     // For now we just grab the first Spark conf.
     if (jobConf == None) {
       jobConf = environmentUpdate.environmentDetails.get("Spark Properties").map(_.toMap)

diff --git a/src/main/scala/com/qubole/sparklens/app/EventHistoryReporter.scala b/src/main/scala/com/qubole/sparklens/app/EventHistoryReporter.scala
@@ -65,6 +65,7 @@ class EventHistoryReporter(file: String, extraConf: List[(String, String)] = Lis
     }
   }
 
+  // TODO Unify with ReporterApp
   private def getFilter(eventString: String): Boolean = {
     implicit val formats = DefaultFormats
     eventFilter.contains(Json4sWrapper.parse(eventString).extract[Map[String, Any]].get("Event")
@@ -81,7 +82,8 @@ class EventHistoryReporter(file: String, extraConf: List[(String, String)] = Lis
       "SparkListenerJobStart",
       "SparkListenerJobEnd",
       "SparkListenerStageSubmitted",
-      "SparkListenerStageCompleted"
+      "SparkListenerStageCompleted",
+      "SparkListenerEnvironmentUpdate",
     )
   }
 

diff --git a/src/main/scala/com/qubole/sparklens/app/ReporterApp.scala b/src/main/scala/com/qubole/sparklens/app/ReporterApp.scala
@@ -116,6 +116,7 @@ object ReporterApp extends App {
     val replayMethod = busKlass.getMethod("replay", classOf[InputStream], classOf[String],
       classOf[Boolean], classOf[(String) => Boolean])
 
+
     replayMethod.invoke(bus, getDecodedInputStream(file, conf), file, boolean2Boolean(false),
       getFilter _)
   }
@@ -139,6 +140,7 @@ object ReporterApp extends App {
 
   }
 
+  // TODO Unify with EventHistoryReporter
   private def getFilter(eventString: String): Boolean = {
     implicit val formats = DefaultFormats
     eventFilter.contains(Json4sWrapper.parse(eventString).extract[Map[String, Any]].get("Event")

diff --git a/src/test/scala/com/qubole/sparklens/app/EventHistoryTest.scala b/src/test/scala/com/qubole/sparklens/app/EventHistoryTest.scala
@@ -48,10 +48,11 @@ class EventHistoryFileReportingSuite extends AnyFunSuite with Matchers {
     fileContents: String,
     appId: String,
     updatedConf: Map[String, String],
-    dynamic: Boolean = false) = {
+    dynamic: Boolean = false,
+    constantExecs: Boolean = true) = {
     fileContents should include(s"ID $appId")
-    if (dynamic) {
-      fileContents should include("i don't remember")
+    if (dynamic && constantExecs) {
+      fileContents should include("Initial execs within tolerance +-10% leaving alone")
     } else {
       fileContents should include("since not dynamic")
     }