title: pySpark Kafka Direct Streaming update Zookeeper / Kafka Offset
body: <p>currently I'm working with Kafka / Zookeeper and pySpark (1.6.0). 
  I have successfully created a kafka consumer, which is using the <code>KafkaUtils.createDirectStream()</code>.</p>
  
  <p>There is no problem with all the streaming, but I recognized, that my Kafka Topics are not updated to the current offset, after I have consumed some messages. </p>
  
  <p>Since we need the topics updated to have a monitoring here in place this is somehow weird. </p>
  
  <p>In the documentation of Spark I found this comment: </p>
  
  <pre><code>   offsetRanges = []
  
       def storeOffsetRanges(rdd):
           global offsetRanges
           offsetRanges = rdd.offsetRanges()
           return rdd
  
       def printOffsetRanges(rdd):
           for o in offsetRanges:
               print "%s %s %s %s" % (o.topic, o.partition, o.fromOffset, o.untilOffset)
  
       directKafkaStream\
           .transform(storeOffsetRanges)\
           .foreachRDD(printOffsetRanges)
  </code></pre>
  
  <blockquote>
    <p>You can use this to update Zookeeper yourself if you want Zookeeper-based Kafka monitoring tools to show progress of the streaming application.</p>
  </blockquote>
  
  <p>Here is the documentation:
  <a href="http://spark.apache.org/docs/1.6.0/streaming-kafka-integration.html#approach-2-direct-approach-no-receivers" rel="nofollow noreferrer">http://spark.apache.org/docs/1.6.0/streaming-kafka-integration.html#approach-2-direct-approach-no-receivers</a></p>
  
  <p>I found a solution in Scala, but I can't find an equivalent for python. 
  Here is the Scala example: <a href="http://geeks.aretotally.in/spark-streaming-kafka-direct-api-store-offsets-in-zk/" rel="nofollow noreferrer">http://geeks.aretotally.in/spark-streaming-kafka-direct-api-store-offsets-in-zk/</a></p>
  
  <h2>Question</h2>
  
  <p>But the question is, how I'm able to update the zookeeper from that point on?</p>