AWS kinesis consumer

kinesis_consumer.py

import  boto
from boto.kinesis.exceptions import ResourceInUseException

import  time

stream_name = "kinesis-fluent-bit-test"

region = 'eu-west-1'

kinesis = boto.kinesis.connect_to_region(region)

r = kinesis.describe_stream(stream_name)
description = r.get('StreamDescription')
status = description.get('StreamStatus')

my_shard_id = description['Shards'][0]['ShardId']


shard_iterator = kinesis.get_shard_iterator(stream_name,my_shard_id,'LATEST')

my_shard_iterator = shard_iterator['ShardIterator']

record_response = kinesis.get_records(my_shard_iterator)
while 'NextShardIterator' in record_response:
    record_response = kinesis.get_records(record_response['NextShardIterator'], 25)

    print(record_response)

    # wait for 5 seconds
    time.sleep(5)