read_offsets.py

def read_offsets(csc, config):
    """Read offset ranges from Cassandra."""
    rows = csc.cassandraTable(config.get_ks(),
                              config.get_offsets_cf())\
        .select("topic", "partition", "from_offset", "until_offset")\
        .where("topic=?", config.get_kafka_topic())\
        .collect()
    offset_ranges = {}
    for r in rows:
        tp = TopicAndPartition(r.topic, r.partition)
        offset_ranges[tp] = r.from_offset
    return offset_ranges if len(offset_ranges) > 0 else None
  
# read in the last successfully processed Kafka offset
from_offsets = read_offsets(cassandra_streaming_context, config)

stream = KafkaUtils.createDirectStream(
   ssc, topics=[config.get_topic()], kafkaParams={
   “metadata.broker.list”: config.get_kafka_brokers(),
   }, fromOffsets=from_offsets)