title: Spark batch reading from Kafka & using Kafka to keep track of offsets
body: <p>I understand that using Kafka's own offset tracking instead of other methods (like checkpointing) is problematic for streaming jobs.</p>
  <p>However I just want to run a Spark batch job every day, reading all messages from the last offset to the most recent and do some ETL with it.</p>
  <p>In theory I want to read this data like so:</p>
  <pre><code>val dataframe = spark.read
        .format(&quot;kafka&quot;)
        .option(&quot;kafka.bootstrap.servers&quot;, &quot;localhost:6001&quot;)
        .option(&quot;subscribe&quot;, &quot;topic-in&quot;)
        .option(&quot;includeHeaders&quot;, &quot;true&quot;)
        .option(&quot;kafka.group.id&quot;, s&quot;consumer-group-for-this-job&quot;)
        .load()
  </code></pre>
  <p>And have Spark commit the offsets back to Kafka based on the <code>group.id</code></p>
  <p>Unfortunately Spark never commits these back, so I went creative and added in the end of my etl job, this code to manually update the offsets for the consumer in Kafka:</p>
  <pre><code>val offsets: Map[TopicPartition, OffsetAndMetadata] = dataFrame
        .select('topic, 'partition, 'offset)
        .groupBy(&quot;topic&quot;, &quot;partition&quot;)
        .agg(max('offset))
        .as[(String, Int, Long)]
        .collect()
        .map {
          case (topic, partition, maxOffset) =&gt; new TopicPartition(topic, partition) -&gt; new OffsetAndMetadata(maxOffset)
        }
        .toMap
  
  val props = new Properties()
      props.put(&quot;group.id&quot;, &quot;consumer-group-for-this-job&quot;)
      props.put(&quot;bootstrap.servers&quot;, &quot;localhost:6001&quot;)
      props.put(&quot;key.deserializer&quot;, &quot;org.apache.kafka.common.serialization.ByteArrayDeserializer&quot;)
      props.put(&quot;value.deserializer&quot;, &quot;org.apache.kafka.common.serialization.ByteArrayDeserializer&quot;)
      props.put(&quot;enable.auto.commit&quot;, &quot;false&quot;)
      val kafkaConsumer = new KafkaConsumer[Array[Byte], Array[Byte]](props)
  
      kafkaConsumer.commitSync(offsets.asJava)
  </code></pre>
  <p>Which technically works, but still next time reading based on this group.id Spark will still start from the beginning.</p>
  <p>Do I have to bite the bullet and keep track of the offsets somewhere, or is there something I'm overlooking?</p>
  <p>BTW I'm testing this with <a href="https://github.com/embeddedkafka/embedded-kafka" rel="nofollow noreferrer">EmbeddedKafka</a></p>