Pyhon操作Kafka數(shù)據(jù)庫
1.Python需要安裝pykafka模塊
pykafka官方手冊： http://pykafka.readthedocs.io/en/latest/index.html
pip install pykafka
2.python 存入kafka消息
# -*- coding: utf-8 -*-
from kafka import KafkaProducer
producer = KafkaProducer(bootstrap_servers=['127.0.0.1:9092'])
#此處ip可以是多個['0.0.0.1:9092','0.0.0.2:9092','0.0.0.3:9092' ]
for i in range(3):
msg = "msg %d" % i
print msg
producer.send('test', msg)
producer.close()
3.讀取本地的所有 topic 信息
from pykafka import KafkaClient
client = KafkaClient(hosts="127.0.0.1:9092")
for topic in client.topics:
print topic
4.查看brokers信息
from pykafka import KafkaClient
client = KafkaClient(hosts="127.0.0.1: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)
5.直接消費kafka
#encoding:utf8
from pykafka import KafkaClient
client = KafkaClient(hosts="127.0.0.1:9092")
topic = client.topics['test']
# 獲取 consumer 消費者
consumer = topic.get_simple_consumer(consumer_group="test",reset_offset_on_start=True)
for message in consumer:
print message
if message is not None:
print ">>>>>>>>>>",message.offset
print ">>>>>>>>>>",message.value
6.從zookeeper消費
from pykafka import KafkaClient
client = KafkaClient(hosts="127.0.0.1:9092")
topic = client.topics['test']
balanced_consumer= topic.get_balanced_consumer(consumer_group='test',auto_commit_enable=True,reset_offset_on_start=True,zookeeper_connect='127.0.0.1:9092')
for message in balanced_consumer:
if message is not None:
print message.offset, message.value
