Super simple AWS Kinesis data streams client in Python

Your Kinesis streams is collecting data and you need a simple and fast way to access those data for testing purpose or a custom processing. You can achieve this in 3 steps:

1. Get the shard id.
2. Get the shard iterator.
3. Get the records.

Follow those steps in any programming language supported by AWS API. Here’s an example written in Python:

import boto3

def process_records(Records):
    #---------------------------
    # Your processing goes here
    #---------------------------
    pass


def main():
    stream_name = 'your-stream-name'
    
    try:
        kinesis_client = boto3.client('kinesis')

        #------------------
        # Get the shard ID.
        #------------------
        response = kinesis_client.describe_stream(StreamName=stream_name)
        shard_id = response['StreamDescription']['Shards'][0]['ShardId']

        #---------------------------------------------------------------------------------------------
        # Get the shard iterator.
        # ShardIteratorType=AT_SEQUENCE_NUMBER|AFTER_SEQUENCE_NUMBER|TRIM_HORIZON|LATEST|AT_TIMESTAMP
        #---------------------------------------------------------------------------------------------
        response = kinesis_client.get_shard_iterator(
            StreamName=stream_name,
            ShardId=shard_id,
            ShardIteratorType='TRIM_HORIZON'
        )
        shard_iterator = response['ShardIterator']

        #-----------------------------------------------------------------
        # Get the records.
        # Get max_records from the shard, or run continuously if you wish.
        #-----------------------------------------------------------------
        max_records = 100
        record_count = 0

        while record_count < max_records:
            response = kinesis_client.get_records(
                ShardIterator=shard_iterator,
                Limit=10
            )
            shard_iterator = response['NextShardIterator']
            records = response['Records']
            record_count += len(records)
            process_records(records)

    except ClientError as e:
        logger.exception("Couldn't get records from stream %s.", stream_name)
        raise


if __name__ == "__main__":
    main()

For more advanced processing, consider using the Kinesis Client Library.

Read every story from Marius Bio (and thousands of other writers on Medium)