kinesis-python 0.1.7

Consumer

The consumer works by launching a process per shard in the stream and then implementing the Python iterator protocol.

from kinesis.consumer import KinesisConsumer

consumer = KinesisConsumer(stream_name='my-stream')
for message in consumer:
    print "Received message: {0}".format(message)

Messages received from each of the shard processes are passed back to the main process through a Python Queue where they are yielded for processing. Messages are not strictly ordered, but this is a property of Kinesis and not this implementation.

from kinesis.consumer import KinesisConsumer
from kinesis.state import DynamoDB

consumer = KinesisConsumer(stream_name='my-stream', state=DynamoDB(table_name='my-kinesis-state'))
for message in consumer:
    print "Received message: {0}".format(message)

Producer

The producer works by launching a single process for accumulation and publishing to the stream.

from kinesis.producer import KinesisProducer

producer = KinesisProducer(stream_name='my-stream')
producer.put('Hello World from Python')

By default the accumulation buffer time is 500ms, or the max record size of 1Mb, whichever occurs first. You can change the buffer time when you instantiate the producer via the buffer_time kwarg, specified in seconds. For example, if your primary concern is budget and not performance you could accumulate over a 60 second duration.

producer = KinesisProducer(stream_name='my-stream', buffer_time=60)

The background process takes precaution to ensure that any accumulated messages are flushed to the stream at shutdown time through signal handlers and the python atexit module, but it is not fully durable and if you were to send a kill -9 to the producer process any accumulated messages would be lost.