Skip to content
Apache Kafka client for Python; high-level & low-level consumer/producer, with great performance.
Branch: master
Clone or download
emmett9001 Merge pull request #927 from carsonip/patch-4
Update README about librdkafka version dependency
Latest commit cfccd8e Mar 18, 2019
Type Name Latest commit message Commit time
Failed to load latest commit information.
benchmark benchmark/: add some documentation / comments Jul 24, 2015
doc update copyright year Jan 7, 2019
pykafka Fix wrong default compacted_topic of ManagedBalancedConsumer Mar 17, 2019
tests Skip PyPy LZ4 test Mar 6, 2019
.coveragerc travis-ci: switch to for coverage reports Oct 19, 2015
.gitignore add python-version to gitignore for pyevn Nov 15, 2018
.travis.yml Bump librdkafka version in travis Feb 23, 2019
CHANGES.rst add code of conduct Nov 16, 2017
CONTRIBUTING.rst update packaging isntructions Sep 18, 2018
LICENSE update licensing Apr 24, 2015
Makefile Fix tests for Apr 26, 2015
README.rst Update README about librdkafka version dependency Mar 17, 2019 keep license from showing in module autodoc Aug 1, 2012
setup.cfg update setup.cfg config block for pytest compatibility Jan 7, 2019 update gevent requirement Oct 12, 2018



PyKafka is a programmer-friendly Kafka client for Python. It includes Python implementations of Kafka producers and consumers, which are optionally backed by a C extension built on librdkafka. It runs under Python 2.7+, Python 3.4+, and PyPy, and supports versions of Kafka 0.8.2 and newer.

PyKafka's primary goal is to provide a similar level of abstraction to the JVM Kafka client using idioms familiar to Python programmers and exposing the most Pythonic API possible.

You can install PyKafka from PyPI with

$ pip install pykafka

or from conda-forge with

$ conda install -c conda-forge pykafka

Full documentation and usage examples for PyKafka can be found on readthedocs.

You can install PyKafka for local development and testing by cloning this repository and running

$ python develop

Getting Started

Assuming you have at least one Kafka instance running on localhost, you can use PyKafka to connect to it.

>>> from pykafka import KafkaClient
>>> client = KafkaClient(hosts=",,...")

Or, for a TLS connection, you might write (and also see SslConfig docs for further details):

>>> from pykafka import KafkaClient, SslConfig
>>> config = SslConfig(cafile='/your/ca.cert',
...                    certfile='/your/client.cert',  # optional
...                    keyfile='/your/client.key',  # optional
...                    password='unlock my client key please')  # optional
>>> client = KafkaClient(hosts="<ssl-port>,...",
...                      ssl_config=config)

If the cluster you've connected to has any topics defined on it, you can list them with:

>>> client.topics
>>> topic = client.topics['my.test']

Once you've got a Topic, you can create a Producer for it and start producing messages.

>>> with topic.get_sync_producer() as producer:
...     for i in range(4):
...         producer.produce('test message ' + str(i ** 2))

The example above would produce to kafka synchronously - the call only returns after we have confirmation that the message made it to the cluster.

To achieve higher throughput, we recommend using the Producer in asynchronous mode, so that produce() calls will return immediately and the producer may opt to send messages in larger batches. The Producer collects produced messages in an internal queue for linger_ms before sending each batch. This delay can be removed or changed at the expense of efficiency with linger_ms, min_queued_messages, and other keyword arguments (see readthedocs). You can still obtain delivery confirmation for messages, through a queue interface which can be enabled by setting delivery_reports=True. Here's a rough usage example:

>>> with topic.get_producer(delivery_reports=True) as producer:
...     count = 0
...     while True:
...         count += 1
...         producer.produce('test msg', partition_key='{}'.format(count))
...         if count % 10 ** 5 == 0:  # adjust this or bring lots of RAM ;)
...             while True:
...                 try:
...                     msg, exc = producer.get_delivery_report(block=False)
...                     if exc is not None:
...                         print 'Failed to deliver msg {}: {}'.format(
...                             msg.partition_key, repr(exc))
...                     else:
...                         print 'Successfully delivered msg {}'.format(
...                         msg.partition_key)
...                 except Queue.Empty:
...                     break

Note that the delivery report queue is thread-local: it will only serve reports for messages which were produced from the current thread. Also, if you're using delivery_reports=True, failing to consume the delivery report queue will cause PyKafka's memory usage to grow unbounded.

You can also consume messages from this topic using a Consumer instance.

>>> consumer = topic.get_simple_consumer()
>>> for message in consumer:
...     if message is not None:
...         print message.offset, message.value
0 test message 0
1 test message 1
2 test message 4
3 test message 9

This SimpleConsumer doesn't scale - if you have two SimpleConsumers consuming the same topic, they will receive duplicate messages. To get around this, you can use the BalancedConsumer.

>>> balanced_consumer = topic.get_balanced_consumer(
...     consumer_group='testgroup',
...     auto_commit_enable=True,
...     zookeeper_connect=','
... )

You can have as many BalancedConsumer instances consuming a topic as that topic has partitions. If they are all connected to the same zookeeper instance, they will communicate with it to automatically balance the partitions between themselves. The partition assignment strategy used by the BalancedConsumer is the "range" strategy by default. The strategy is switchable via the membership_protocol keyword argument, and can be either an object exposed by pykafka.membershipprotocol or a custom instance of pykafka.membershipprotocol.GroupMembershipProtocol.

You can also use the Kafka 0.9 Group Membership API with the managed keyword argument on get_balanced_consumer.

Using the librdkafka extension

PyKafka includes a C extension that makes use of librdkafka to speed up producer and consumer operation. PyKafka requires librdkafka v0.9.1+. Some system package managers may not have up-to-date versions. To use the librdkafka extension, you need to make sure the header files and shared library are somewhere where python can find them, both when you build the extension (which is taken care of by develop) and at run time. Typically, this means that you need to either install librdkafka in a place conventional for your system, or declare C_INCLUDE_PATH, LIBRARY_PATH, and LD_LIBRARY_PATH in your shell environment to point to the installation location of the librdkafka shared objects. You can find this location with locate

After that, all that's needed is that you pass an extra parameter use_rdkafka=True to topic.get_producer(), topic.get_simple_consumer(), or topic.get_balanced_consumer(). Note that some configuration options may have different optimal values; it may be worthwhile to consult librdkafka's configuration notes for this.

Operational Tools

PyKafka includes a small collection of CLI tools that can help with common tasks related to the administration of a Kafka cluster, including offset and lag monitoring and topic inspection. The full, up-to-date interface for these tools can be fould by running

$ python cli/ --help

or after installing PyKafka via setuptools or pip:

$ kafka-tools --help

PyKafka or kafka-python?

These are two different projects. See the discussion here for comparisons between the two projects.


If you're interested in contributing code to PyKafka, a good place to start is the "help wanted" issue tag. We also recommend taking a look at the contribution guide.


If you need help using PyKafka, there are a bunch of resources available. For usage questions or common recipes, check out the StackOverflow tag. The Google Group can be useful for more in-depth questions or inquries you'd like to send directly to the PyKafka maintainers. If you believe you've found a bug in PyKafka, please open a github issue after reading the contribution guide.

You can’t perform that action at this time.
You signed in with another tab or window. Reload to refresh your session. You signed out in another tab or window. Reload to refresh your session.