Skip to main content

Low level, multiprocessing based AWS Kinesis producer & consumer library

Project description

https://img.shields.io/travis/NerdWalletOSS/kinesis-python.svg https://img.shields.io/codecov/c/github/NerdWalletOSS/kinesis-python.svg Latest PyPI version

The official Kinesis python library requires the use of Amazon’s “MultiLangDaemon”, which is a Java executable that operates by piping messages over STDIN/STDOUT.

ಠ_ಠ

While the desire to have a single implementation of the client library from a maintenance standpoint makes sense for the team responsible for the KPL, requiring the JRE to be installed and having to account for the overhead of the stream being consumed by Java and Python is not desireable for teams working in environments without Java.

This is a pure-Python implementation of Kinesis producer and consumer classes that leverages Python’s multiprocessing module to spawn a process per shard and then sends the messages back to the main process via a Queue. It only depends on boto3 (AWS SDK), offspring (Subprocess implementation) and six (py2/py3 compatibility).

It also includes a DynamoDB state back-end that allows for multi-instance consumption of multiple shards, and stores the checkpoint data so that you can resume where you left off in a stream following restarts or crashes.

Overview

All of the functionality is wrapped in two classes: KinesisConsumer and KinesisProducer

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.

Locking, Checkpointing & Multi-instance consumption

When deploying an application with multiple instances DynamoDB can be leveraged as a way to coordinate which instance is responsible for which shard, as it is not desirable to have each instance process all records.

With or without multiple nodes it is also desirable to checkpoint the stream as you process records so that you can pickup from where you left off if you restart the consumer.

A “state” backend that leverages DynamoDB allows consumers to coordinate which node is responsible which shards and where in the stream we are currently reading from.

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)

The DynamoDB table must already exist and must have a HASH key of shard, with type S (string).

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.

AWS Permissions

By default the producer, consumer & state classes all use the default boto3 credentials chain. If you wish to alter this you can instantiate your own boto3.Session object and pass it into the constructor via the boto3_session keyword argument of KinesisProducer, KinesisConsumer or DynamoDB.

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

kinesis-python-0.2.1.tar.gz (11.7 kB view details)

Uploaded Source

Built Distribution

kinesis_python-0.2.1-py3-none-any.whl (10.7 kB view details)

Uploaded Python 3

File details

Details for the file kinesis-python-0.2.1.tar.gz.

File metadata

  • Download URL: kinesis-python-0.2.1.tar.gz
  • Upload date:
  • Size: 11.7 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/1.14.0 pkginfo/1.5.0.1 requests/2.22.0 setuptools/41.2.0 requests-toolbelt/0.9.1 tqdm/4.35.0 CPython/3.7.1

File hashes

Hashes for kinesis-python-0.2.1.tar.gz
Algorithm Hash digest
SHA256 4f3e98f337b0d5e5c77b26a9138e627da32104ade3785b3a136c70389c2b79d3
MD5 1394f9c63f360569da6e4d67e14e5368
BLAKE2b-256 bebb7e54b23b64fd49ad9afedd5f3af992980a744023ab01b963e99ea2572a61

See more details on using hashes here.

File details

Details for the file kinesis_python-0.2.1-py3-none-any.whl.

File metadata

  • Download URL: kinesis_python-0.2.1-py3-none-any.whl
  • Upload date:
  • Size: 10.7 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/1.14.0 pkginfo/1.5.0.1 requests/2.22.0 setuptools/41.2.0 requests-toolbelt/0.9.1 tqdm/4.35.0 CPython/3.7.1

File hashes

Hashes for kinesis_python-0.2.1-py3-none-any.whl
Algorithm Hash digest
SHA256 73aeeffc1f48bdba7a3390939f3b96465145772edbcf96daf2f8c41071f6c48d
MD5 0ef46ad131bb754c5bcef11173ac8f09
BLAKE2b-256 9d560a82a502f8bd438ee1d24a5561dac323e10e0c503918e3245e241cd720c9

See more details on using hashes here.

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page