首先要安装连接kafka的库,kafka-python和pykafka都可以,但是kafka-python好像是不支持zk连接,所以就使用pykafka了
pip install pykafka
1.查看所有topics:
from pykafka import KafkaClient client = KafkaClient(hosts="192.168.8.101:9092,192.168.8.102:9092,192.168.8.103:9092,192.168.8.104:9092") print client.topics
2.查看所有brokers信息
from pykafka import KafkaClient
client = KafkaClient(hosts="192.168.8.101:9092,192.168.8.102:9092,192.168.8.103:9092,192.168.8.104:9092")
#print client.brokers
for n in client.brokers:
host = client.brokers[n].host
port = client.brokers[n].port
id = client.brokers[n].id
print "host=%s | port=%s | broker.id=%s " %(host,port,id)
3.直接消费kafka
from pykafka import KafkaClient
client = KafkaClient(hosts="192.168.8.101:9092,192.168.8.102:9092,192.168.8.103:9092,192.168.8.104:9092")
topic = client.topics['linuxhub']
consumer = topic.get_simple_consumer(
consumer_group="linuxhub",
# auto_offset_reset=OffsetType.EARLIEST,
reset_offset_on_start=True
)
for message in consumer:
if message is not None:
print message.offset, message.value
4.从zk消费kafka
from pykafka import KafkaClient
client = KafkaClient(hosts="192.168.8.101:9092,192.168.8.102:9092,192.168.8.103:9092,192.168.8.104:9092")
topic = client.topics['linuxhub']
balanced_consumer= topic.get_balanced_consumer(
consumer_group='linuxhub',
auto_commit_enable=True,
# reset_offset_on_start=True,
zookeeper_connect='192.168.8.101:2181,192.168.8.102:2181,192.168.8.103:2181'
)
for message in balanced_consumer:
if message is not None:
print message.offset, message.value