save_offsets.py

def save_offset_ranges(csc, config, rdd):
    """Save offset ranges for a given RDD to Cassandra."""
    for o in rdd.offsetRanges():
        save_offset_to_cassandra(csc, config, o)
        
def save_offset_to_cassandra(csc, config, offset):
    """Save offset range to Cassandra."""
    rdd = csc.parallelize([{
        "topic": offset.topic,
        "partition": offset.partition,
        "from_offset": offset.fromOffset,
        "until_offset": offset.untilOffset
    }])
    rdd.saveToCassandra(config.get_ks(), config.get_offsets_cf())
    
# save the latest Kafka offset ranges
kms.foreachRDD(lambda x: (save_offset_ranges(job.csc, job.get_config(), x)))