Reading a checkpointed RDD

Viewed 745

So I'm attempting to pre-compute my data such that I can pull an RDD from a checkpoint. However, I'm finding that upon running the same job twice the system is simply recreating the RDD from scratch.

Here is the code I'm implementing to create the checkpoint:

  def checkpointTeam(checkpointDir:String) ={
    sparkContext.setCheckpointDir(checkpointDir)
    val a = ObjectsTable.readAsRDD(sparkContext, Some("5507424...")).persist() //read from HBase
    a.checkpoint()
    a.count()
  }

edit: here is the code that creates my sparkContext

val sparkContext = {
      val sparkConf = new SparkConf()
        .setAppName(appName.getOrElse(jobName))

        .set("spark.driver.allowMultipleContexts", "true")


      sparkConf.setMaster(EnvConfig.sparkMaster)

      Option(System.getenv("SPARK_HOME")).foreach(sparkConf.setSparkHome)

      sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")    
      sparkConf.set("spark.scheduler.mode", "FAIR")
      sparkConf.set("spark.yarn.am.memory", "14g")
      sparkConf.set("spark.yarn.am.cores", "4")
      sparkConf.set("spark.driver.extraJavaOptions","-XX:+UseConcMarkSweepGC -XX:+CMSIncrementalMode")
      sparkConf.set("spark.executor.extraJavaOptions","-XX:+UseConcMarkSweepGC -XX:+CMSIncrementalMode")
      sparkConf.set("spark.storage.memoryFraction", "0.4")
      sparkConf.set("spark.yarn.executor.memoryOverhead", "4000")

      sparkConf.set("spark.locality.wait.process","120000")

      new SparkContext(sparkConf)
    }

    sparkContext.addSparkListener(new StatsReportListener())

    log.info("Initialized SparkContext")

    sparkContext
  }
1 Answers
Related