Can SparkContext.setCheckpointDir(hdfsPath) set same hdfsPath in different spark apps?

user_4978515 0
Viewed 3297

As docs of:

https://spark.apache.org/docs/2.2.1/api/java/org/apache/spark/SparkContext.html#setCheckpointDir-java.lang.String-

SparkContext:

setCheckpointDir
public void setCheckpointDir(String directory)
Set the directory under which RDDs are going to be checkpointed.
Parameters:
directory - path to the directory where checkpoint files will be stored (must be HDFS path if running in cluster)

Questions : 1) If different spark apps SparkContext.setCheckpointDir(hdfsPath) set the same hdfsPath, Is there any conflict?

2) If no conflict, the hdfsPath for CheckpointDir will clean automaticly?

apache-spark scala
1 Answers

Questions :

1) If different spark apps SparkContext.setCheckpointDir(hdfsPath) set the same hdfsPath, Is there any conflict?

Answer : No conflict as per below example given. Multiple applcaition can use same check point directory. Under that unique hash kind of folder will be created to avoid conflicts.

2) If no conflict, the hdfsPath for CheckpointDir will clean automaticly?

Answer : Yes its happening. for the below example I used local for demonstration... but local or hdfs it doesnt matter. Behaviour will be the same.

Lets go by example (ran multiple times with same check point directory): 

package examples

import java.io.File

import org.apache.log4j.Level


object CheckPointTest extends App {
  import org.apache.spark.sql.{Dataset, SparkSession}
  val spark = SparkSession.builder().appName("CheckPointTest").master("local").getOrCreate()
  val logger = org.apache.log4j.Logger.getLogger("org")
  logger.setLevel(Level.WARN)
  import spark.implicits._

  spark.sparkContext.setCheckpointDir("/tmp/checkpoints")


  val csvData1: Dataset[String] = spark.sparkContext.parallelize(
    """
      |id
      | a
      | b
      | c

    """.stripMargin.lines.toList).toDS()
  val frame1 = spark.read.option("header", true).option("inferSchema",true).csv(csvData1).show

  val checkpointDir = spark.sparkContext.getCheckpointDir.get
  println(checkpointDir)

println("Number of Files in Check Point Directory " + getListOfFiles(checkpointDir).length)


  def getListOfFiles(dir: String):List[File] = {
    val d = new File(dir)
    if (d.exists && d.isDirectory) {
      d.listFiles.filter(_.isFile).toList
    } else {
      List[File]()
    }
  }
}

Result : 

+---+
| id|
+---+
|  a|
|  b|
|  c|
+---+

file:/tmp/checkpoints/30e6f882-b49a-42cc-9e60-59adecf13166
Number of Files in Check Point Directory 0 // this indicates once application finished removed all the RDD/DS information.

If you have a look at checkpoint folder it will be like this...


user@f0189843ecbe [~/Downloads]$ ll /tmp/checkpoints/
total 0
drwxr-xr-x  2 user  wheel   64 Mar 27 14:08 a2396c08-14b6-418a-b183-a90a4ca7dba3
drwxr-xr-x  2 user  wheel   64 Mar 27 14:09 65c8ef5a-0e64-4e79-a050-7d1ee1d0e03d
drwxr-xr-x  2 user  wheel   64 Mar 27 14:09 5667758c-180f-4c0b-8b3c-912afca59f55
drwxr-xr-x  2 user  wheel   64 Mar 27 14:10 30e6f882-b49a-42cc-9e60-59adecf13166
drwxr-xr-x  6 user  wheel  192 Mar 27 14:10 .
drwxrwxrwt  5 root    wheel  160 Mar 27 14:10 ..
user@f0189843ecbe [~/Downloads]$ du -h /tmp/checkpoints/
  0B    /tmp/checkpoints//a2396c08-14b6-418a-b183-a90a4ca7dba3
  0B    /tmp/checkpoints//5667758c-180f-4c0b-8b3c-912afca59f55
  0B    /tmp/checkpoints//65c8ef5a-0e64-4e79-a050-7d1ee1d0e03d
  0B    /tmp/checkpoints//30e6f882-b49a-42cc-9e60-59adecf13166
  0B    /tmp/checkpoints/

Conclusion :

1) Even multiple applications are running parllel, there will be unique hash under check point directory in that all the RDD/DS information will be stored.

2) Afer success full execution of each Spark Application, the context cleaner will remove the contents in it.. is what I observed from the above practical example.

user_6470530
Related