python kafka client--confluent-kafka-python

項目中需要使用python 向Kafka生產和消費數據，最初使用pykafka 。后來發現pykafka不支持client.id.
最后，終於找到confluent-kafka。

python kafka推薦使用confluent-kafka，官方推薦的。

Confluent's Apache Kafka Python client
confluent-kafka-python is Confluent's Python client for Apache Kafka and the Confluent Platform.

安裝和使用可參考github
https://github.com/confluentinc/confluent-kafka-python

本文只針對筆者使用python2.7 + Linux環境介紹kafka包安裝和使用。

安裝python
可網上搜索教程，此處略過。

安裝librdkafka

For RedHat and RPM-based distros, add this YUM repo and then do sudo yum install librdkafka-devel python-devel: http://docs.confluent.io/current/installation.html#rpm-packages-via-yum

其他系統，可以參考github上說明。

安裝 python kafka pacakge

$ pip install confluent-kafka

Successfully installed confluent-kafka-0.11.4

consumer 舉例

\#!/usr/bin/python

from confluent_kafka import Consumer, KafkaError

mybroker = "127.0.0.1:9092"

c = Consumer({
    'bootstrap.servers': mybroker,
    'group.id': 'mygroup',
    'client.id': 'lanyang',
    'default.topic.config': {
    'auto.offset.reset': 'smallest'
    }
})

c.subscribe(['test'])

while True:
    msg = c.poll(1.0)

    if msg is None:
        continue
    if msg.error():
        if msg.error().code() == KafkaError._PARTITION_EOF:
            continue
        else:
            print(msg.error())
            break

    print('Received message: {}'.format(msg.value().decode('utf-8')))

c.close()

輸出結果

$ python kafa_consumer.py 
Received message: Hello,python
Received message: Hello,kafka

更多demo請參考github。