SparkListener - Checkpointing jobs

sparklistener-checkpointing.scala

import scala.collection.JavaConversions._    // for propertiesAsScalaMap function

sc.addSparkListener(new SparkListener() {
override def onJobStart(jobStart: SparkListenerJobStart) {
  println("ADAPT: INSIDE Job Start Listener ");
  var props = propertiesAsScalaMap(jobStart.properties)
  if (props.contains("spark.rdd.scope"))
  {
      val propsMap = // Convert props to propsMap
      if ( propsMap.contains("name") && propsMap("name") == "checkpoint")
      {
        println("JobID " + jobStart.jobId);
        println("This is a checkpointing job for RDD - " + propsMap("id"))
        println("StartTime - " + jobStart.time)
      }
  }
}

override def onJobEnd(jobEnd: SparkListenerJobEnd) {
  println("ADAPT: Inside Job end Listener ");
  println("JobID " + jobEnd.jobId);
  println("EndTime " + jobEnd.time);
}

// There are listeners for other events too, but not required for now. Below are some examples

/*
override def onApplicationStart(applicationStart: SparkListenerApplicationStart) {
  println("ADAPT: Spark ApplicationStart: " + applicationStart.appName + " at time - " + applicationStart.time);
}

override def onApplicationEnd(applicationEnd: SparkListenerApplicationEnd) {
  println("ADAPT: Spark ApplicationEnd: " + applicationEnd.time);
}

override def onStageCompleted(stageCompleted: SparkListenerStageCompleted) {
  println("ADAPT: Inside Stage Completed Listener ");
  val stageInfo = stageCompleted.stageInfo;
  val rddInfos = stageInfo.rddInfos;
  println(" The rddInfos for this stage are - ");
  rddInfos.foreach {println}
  for (rddInfo <- rddInfos) println(rddInfo.id)

  rddInfos.foreach(row => {
      println("ADAPT: Listener - rdd memSize " + row.memSize)
      println("ADAPT: Listener - rdd diskSize " + row.diskSize)
    })

  println("\n");
}
*/

});