Repository files navigation

confluent-kafka-python provides a high-level Producer, Consumer and AdminClient compatible with allApache Kafka^TM brokers >= v0.8,Confluent CloudandConfluent Platform. The client is:

• Reliable - It's a wrapper aroundlibrdkafka (provided automatically via binary wheels) which is widely deployed in a diverse set of production scenarios. It's tested usingthe same set of system tests as the Java clientand more. It's supported byConfluent.

• Performant - Performance is a key design consideration. Maximum throughput is on par with the Java client for larger message sizes (where the overhead of the Python interpreter has less impact). Latency is on par with the Java client.

• Future proof - Confluent, founded by thecreators of Kafka, is building astreaming platformwith Apache Kafka at its core. It's high priority for us that client features keeppace with core Apache Kafka and components of theConfluent Platform.

For a step-by-step guide on using the client seeGetting Started with Apache Kafka and Python.

Aditional examples can be found in theexamples directory or theconfluentinc/examples github repo, which include demonstration of:

• Exactly once data processing using the transactional API.
• Integration with asyncio.
• (De)serializing Protobuf, JSON, and Avro data with Confluent Schema Registry integration.
• Confluent Cloud configuration.

Also refer to theAPI documentation.

Finally, thetests are useful as a reference for example usage.

fromconfluent_kafkaimportProducerp=Producer({'bootstrap.servers':'mybroker1,mybroker2'})defdelivery_report(err,msg):""" Called once for each message produced to indicate delivery result.        Triggered by poll() or flush(). """iferrisnotNone:print('Message delivery failed: {}'.format(err))else:print('Message delivered to {} [{}]'.format(msg.topic(),msg.partition()))fordatainsome_data_source:# Trigger any available delivery report callbacks from previous produce() callsp.poll(0)# Asynchronously produce a message. The delivery report callback will# be triggered from the call to poll() above, or flush() below, when the# message has been successfully delivered or failed permanently.p.produce('mytopic',data.encode('utf-8'),callback=delivery_report)# Wait for any outstanding messages to be delivered and delivery report# callbacks to be triggered.p.flush()

For a discussion on the poll based producer API, refer to theIntegrating Apache Kafka With Python Asyncio Web Applicationsblog post.

fromconfluent_kafkaimportConsumerc=Consumer({'bootstrap.servers':'mybroker','group.id':'mygroup','auto.offset.reset':'earliest'})c.subscribe(['mytopic'])whileTrue:msg=c.poll(1.0)ifmsgisNone:continueifmsg.error():print("Consumer error: {}".format(msg.error()))continueprint('Received message: {}'.format(msg.value().decode('utf-8')))c.close()

Create topics:

fromconfluent_kafka.adminimportAdminClient,NewTopica=AdminClient({'bootstrap.servers':'mybroker'})new_topics= [NewTopic(topic,num_partitions=3,replication_factor=1)fortopicin ["topic1","topic2"]]# Note: In a multi-cluster production scenario, it is more typical to use a replication_factor of 3 for durability.# Call create_topics to asynchronously create topics. A dict# of <topic,future> is returned.fs=a.create_topics(new_topics)# Wait for each operation to finish.fortopic,finfs.items():try:f.result()# The result itself is Noneprint("Topic {} created".format(topic))exceptExceptionase:print("Failed to create topic {}: {}".format(topic,e))

TheProducer,Consumer andAdminClient are all thread safe.

Install self-contained binary wheels

$ pip install confluent-kafka

NOTE: The pre-built Linux wheels do NOT contain SASL Kerberos/GSSAPI support.If you need SASL Kerberos/GSSAPI support you must install librdkafka andits dependencies using the repositories below and then buildconfluent-kafka using the instructions in the"Install from source" section below.

To use Schema Registry with the Avro serializer/deserializer:

$ pip install "confluent-kafka[avro,schemaregistry]"

To use Schema Registry with the JSON serializer/deserializer:

$ pip install "confluent-kafka[json,schemaregistry]"

To use Schema Registry with the Protobuf serializer/deserializer:

$ pip install "confluent-kafka[protobuf,schemaregistry]"

When using Data Contract rules (including CSFLE) add therulesextra, e.g.:

$ pip install "confluent-kafka[avro,schemaregistry,rules]"

Install from source

For source install, see theInstall from source section inINSTALL.md.

The Python client (as well as the underlying C library librdkafka) supportsall broker versions >= 0.8.But due to the nature of the Kafka protocol in broker versions 0.8 and 0.9 itis not safe for a client to assume what protocol version is actually supportedby the broker, thus you will need to hint the Python client what protocolversion it may use. This is done through two configuration settings:

• broker.version.fallback=YOUR_BROKER_VERSION (default 0.9.0.1)
• api.version.request=true|false (default true)

When using a Kafka 0.10 broker or later you don't need to do anything(api.version.request=true is the default).If you use Kafka broker 0.9 or 0.8 you must setapi.version.request=false and setbroker.version.fallback to your broker version,e.gbroker.version.fallback=0.9.0.1.

More info here:https://github.com/edenhill/librdkafka/wiki/Broker-version-compatibility

If you're connecting to a Kafka cluster through SSL you will need to configurethe client with'security.protocol': 'SSL' (or'SASL_SSL' if SASLauthentication is used).

The client will use CA certificates to verify the broker's certificate.The embedded OpenSSL library will look for CA certificates in/usr/lib/ssl/certs/or/usr/lib/ssl/cacert.pem. CA certificates are typically provided by theLinux distribution'sca-certificates package which needs to be installedthroughapt,yum, et.al.

If your system stores CA certificates in another location you will need toconfigure the client with'ssl.ca.location': '/path/to/cacert.pem'.

Alternatively, the CA certificates can be provided by thecertifiPython package. To use certifi, add animport certifi line and configure theclient's CA location with'ssl.ca.location': certifi.where().

Apache License v2.0

KAFKA is a registered trademark of The Apache Software Foundation and has been licensed for useby confluent-kafka-python. confluent-kafka-python has no affiliation with and is not endorsed byThe Apache Software Foundation.

Instructions on building and testing confluent-kafka-python can be foundhere.

For a step-by-step guide on using the Python client with Confluent Cloud seeGetting Started with Apache Kafka and Python onConfluent Developer.