title: Spark Structured Streaming checkpoint usage in production
body: <p>I have troubles understanding how checkpoints work when working with Spark Structured streaming.</p>
  <p>I have a spark process that generates some events, which I log in an Hive table.
  For those events, I receive a confirmation event in a kafka stream.</p>
  <p>I created a new spark process that</p>
  <ul>
  <li>reads the events from the Hive log table into a DataFrame</li>
  <li>joins those events with the stream of confirmation events using Spark Structured Streaming</li>
  <li>writes the joined DataFrame to an HBase table.</li>
  </ul>
  <p>I tested the code in spark-shell and it works fine, below the pseudocode (I'm using Scala).</p>
  <pre><code>val tableA = spark.table(&quot;tableA&quot;)
  
  val startingOffset = &quot;earliest&quot;
  
  val streamOfData = .readStream 
    .format(&quot;kafka&quot;) 
    .option(&quot;startingOffsets&quot;, startingOffsets)
    .option(&quot;otherOptions&quot;, otherOptions)
  
  val joinTableAWithStreamOfData = streamOfData.join(tableA, Seq(&quot;a&quot;), &quot;inner&quot;)
  
  joinTableAWithStreamOfData 
    .writeStream
    .foreach(
      writeDataToHBaseTable()
    ).start()
    .awaitTermination()
  
  </code></pre>
  <p>Now I would like to schedule this code to run periodically, e.g. every 15 minutes, and I'm struggling understanding <strong>how to use checkpoints</strong> here.</p>
  <p>At every run of this code, I would like to <strong>read from the stream only the events I haven't read yet in the previous run</strong>, and inner join those new events with my log table, so to write only new data to the final HBase table.</p>
  <p>I created a directory in HDFS where to store the checkpoint file.
  I provided that location to the spark-submit command I use to call the spark code.</p>
  <pre><code>spark-submit --conf spark.sql.streaming.checkpointLocation=path_to_hdfs_checkpoint_directory 
  --all_the_other_settings_and_libraries
  </code></pre>
  <p>At this moment the code runs fine every 15 minutes without any error, but it doesn't do anything basically since it is not dumping the new events to the HBase table.
  Also the checkpoint directory is empty, while I assume some file has to be written there?</p>
  <p>And does the readStream function need to be adapted so to start reading from the latest checkpoint?</p>
  <pre><code>val streamOfData = .readStream 
    .format(&quot;kafka&quot;) 
    .option(&quot;startingOffsets&quot;, startingOffsets) ??
    .option(&quot;otherOptions&quot;, otherOptions)
  </code></pre>
  <p>I'm really struggling to understand the spark documentation regarding this.</p>
  <p>Thank you in advance!</p>