title: How to get Kafka offsets for structured query for manual and reliable offset management?
body: <p>Spark 2.2 introduced a Kafka's structured streaming source. As I understand, it's relying on HDFS checkpoint directory to store offsets and guarantee an "exactly-once" message delivery.</p>
  
  <p>But old docks (like <a href="https://blog.cloudera.com/blog/2017/06/offset-management-for-apache-kafka-with-apache-spark-streaming/" rel="noreferrer">https://blog.cloudera.com/blog/2017/06/offset-management-for-apache-kafka-with-apache-spark-streaming/</a>) says that  Spark Streaming checkpoints are not recoverable across applications or Spark upgrades and hence not very reliable. As a solution, there is a practice to support storing offsets in external storage that supports transactions like MySQL or RedshiftDB.</p>
  
  <p>If I want to store offsets from Kafka source to a transactional DB, how can I obtain offset from a structured stream batch? </p>
  
  <p>Previously, it can be done by casting RDD to <code>HasOffsetRanges</code>:</p>
  
  <pre class="lang-scala prettyprint-override"><code>val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges    
  </code></pre>
  
  <p>But with new Streaming API, I have an <code>Dataset</code> of <code>InternalRow</code> and I can't find an easy way to fetch offsets. The Sink API has only <code>addBatch(batchId: Long, data: DataFrame)</code> method and how can I suppose to get an offset for given batch id? </p>