Может ли SparkContext.setCheckpointDir (hdfsPath) установить один и тот же hdfsPath в разных приложениях spark?

Question

Как документы:

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)

---

Вопросы: 1) Если разные приложения spark SparkContext.setCheckpointDir(hdfsPath) устанавливают один и тот же hdfsPath, есть ли конфликт?

2) Если нет конфликта, hdfsPath для CheckpointDir будет очищаться автоматически

Answer 1

Вопросы:

1) Если разные Spark-приложения SparkContext.setCheckpointDir (hdfsPath) устанавливают один и тот же hdfsPath, есть ли конфликт?

Ответ: Нет конфликта в соответствии с приведенным ниже примером. Несколько приложений могут использовать один и тот же каталог контрольных точек. Под этим уникальным типом папки ha sh будет создан, чтобы избежать конфликтов.

---

2) Если нет конфликта, hdfsPath для CheckpointDir будет очищать автоматически?

Ответ : Да это происходит. для примера ниже я использовал local для демонстрации ... но local или hdfs это не имеет значения. Поведение будет таким же.

---

Позволяет go на примере (запускается несколько раз с одним и тем же каталогом контрольных точек):

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]()
    }
  }
}

Результат:

+---+
| 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.

Если вы посмотрите на папку контрольных точек, она будет выглядеть следующим образом ...

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/

Вывод:

1) Даже если несколько приложений работают parllel, будет уникальный ha sh в каталоге контрольных точек, в котором будет храниться вся информация RDD / DS.

2) После успешного полного выполнения каждого Spark-приложения очиститель контекста удалит в нем содержимое. Это то, что я наблюдал из приведенного выше практического примера.