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
}