spark-streaming-hdfs-marker-file.scala

  val checkpointDirectory = "hdfs:///path/to/checkpoint/dir"

  def main(args: Array[String]) {

    // Get StreamingContext from checkpoint data or create a new one
    val context = StreamingContext.getOrCreate(checkpointDirectory, functionToCreateContext _)

    // Start the context
    context.start()

    // check every 10s for shutdown hdfs file
    val checkIntervalMillis = 10000
    var isStopped = false
    while (!isStopped) {
      val isStopped = context.awaitTerminationOrTimeout(checkIntervalMillis)
      if (!isStopped && isShutdownRequested) {
        val stopSparkContext = true
        val stopGracefully = true
        context.stop(stopSparkContext, stopGracefully)
      }
    }
  }

  def isShutdownRequested(): Boolean = {
    val fs = FileSystem.get(new Configuration())
    fs.exists(new Path("hdfs:///path/to/shutdown/file"))
  }

  // Function to create and setup a new StreamingContext
  def functionToCreateContext(): StreamingContext = {
    val sparkConf = new SparkConf()
    val batchDuration = Durations.seconds(5)
    val ssc = new StreamingContext(sparkConf, batchDuration)
    // ... create DStream and define transformations and actions
    ssc.checkpoint(checkpointDirectory) // set checkpoint directory
    ssc
  }