[toc]

安装

pykafka github

$ pip install pykafka$ conda install -c conda-forge pykafka注意kafka版本只支持 kafka 1.1, 1.0,0.11, 0.10, 0.9,0.8 （201902）该作者在https://github.com/dpkp/kafka-python/pull/1152 这个推送增加了kerberos支持

验证kerberos

kerberos知识

配置一般在consumer.properties中
拆解一个Principal：
xianglei/dmp-master1.hadoop@HADOOP.COM
一个完整的Principal由3个部分构成。用户名/FQDN（Full Quafilied Domain Name）的主机名@REALM（受保护的域，全大写）当然这个用户名需要是Linux下存在的用户FQDN全限定域名，就是一定要带上hostname.domain这种形式，当然，如果你的主机并没有给出domain，那么不写域名也可以。反正就是要全部的主机名加域名（如果存在域名的话）。但实际上，在Kerberos里面，这个并不称之为主机名，而是叫做Instance，实例名，他可以不是任何服务器的主机名称，但是便于理解和认识，我们还是先把他当初主机名来看待吧。REALM，受到Kerberos保护的域名称，就是一类或一组受到Kerberos保护服务的服务器集合，你可以想象成Windows里面的域。由于一个KDC可以同时保护多个域，比如你可以在一个KDC上既保护HADOOP服务器组，也保护MYSQL服务器组，所以我们通常会使用域名来进行区别。如果你的hostname里面使用了domain name，那么你必须在Principal的第二部分写完整，否则KDC将无法验证主机的合法性，加密的tgt是要带着主机名信息的。还有，特别需要注意的是，这里面第二部分的domain（域名），第三部分的realm（域），在中文里的字是一样，但是英文单词完全不同，他们所表达的含义也完全不同。由于通常Kerberos的Realm部分也会写成域名的形式，所以就会让人迷惑，而实际上，你把realm部分理解成windows里面的workgroup或者home这种域也是可以的。名称可以随便起，不一定用你的真实域名。只是个区分不同服务集合的代号。

使用

资料

我是用来连接华为kafka的，测试可以通过kerberos验证。具体代码就不贴了，引用一下其他作者的，感谢＃coding=utf8from kafka import KafkaProducerproducer = KafkaProducer(bootstrap_servers=["xxxx:9200"],security_protocol="SASL_PLAINTEXT",sasl_mechanism="GSSAPI",sasl_kerberos_service_name="kafka")
print "connect success."
future = producer.send("xxxx", "test")
result = future.get(timeout=60)
print "send success."

其他示例代码

原贴

kafka简介（摘自百度百科）一、简介：
详见：https://blog.csdn.net/Beyond_F4/article/details/80310507二、安装
详见博客：https://blog.csdn.net/beyond_f4/article/details/80095689              三、按照官网的样例，先跑一个应用
1、生产者:
from kafka import KafkaProducerproducer = KafkaProducer(bootstrap_servers=['172.21.10.136: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" % iproducer.send('test', msg)
producer.close()2、消费者(简单demo):
from kafka import KafkaConsumerconsumer = KafkaConsumer('test',bootstrap_servers=['172.21.10.136:9092'])for message in consumer:print ("%s:%d:%d: key=%s value=%s" % (message.topic, message.partition,message.offset, message.key,message.value))启动后生产者、消费者可以正常消费。3、消费者(消费群组)
from kafka import KafkaConsumerconsumer = KafkaConsumer('test',group_id='my-group',bootstrap_servers=['172.21.10.136:9092'])for message in consumer:print ("%s:%d:%d: key=%s value=%s" % (message.topic, message.partition,message.offset, message.key,message.value))启动多个消费者，只有其中可以可以消费到，满足要求，消费组可以横向扩展提高处理能力4、消费者(读取目前最早可读的消息)
from kafka import KafkaConsumerconsumer = KafkaConsumer('test',auto_offset_reset='earliest',bootstrap_servers=['172.21.10.136:9092'])for message in consumer:print ("%s:%d:%d: key=%s value=%s" % (message.topic, message.partition,message.offset, message.key,message.value))auto_offset_reset:重置偏移量，earliest移到最早的可用消息，latest最新的消息，默认为latest
源码定义:{'smallest': 'earliest', 'largest': 'latest'}5、消费者(手动设置偏移量)
from kafka import KafkaConsumer
from kafka.structs import TopicPartitionconsumer = KafkaConsumer('test',bootstrap_servers=['172.21.10.136:9092'])print consumer.partitions_for_topic("test")  #获取test主题的分区信息
print consumer.topics()  #获取主题列表
print consumer.subscription()  #获取当前消费者订阅的主题
print consumer.assignment()  #获取当前消费者topic、分区信息
print consumer.beginning_offsets(consumer.assignment()) #获取当前消费者可消费的偏移量
consumer.seek(TopicPartition(topic=u'test', partition=0), 5)  #重置偏移量，从第5个偏移量消费
for message in consumer:print ("%s:%d:%d: key=%s value=%s" % (message.topic, message.partition,message.offset, message.key,message.value))6、消费者(订阅多个主题)
from kafka import KafkaConsumer
from kafka.structs import TopicPartitionconsumer = KafkaConsumer(bootstrap_servers=['172.21.10.136:9092'])
consumer.subscribe(topics=('test','test0'))  #订阅要消费的主题
print consumer.topics()
print consumer.position(TopicPartition(topic=u'test', partition=0)) #获取当前主题的最新偏移量
for message in consumer:print ("%s:%d:%d: key=%s value=%s" % (message.topic, message.partition,message.offset, message.key,message.value))7、消费者(手动拉取消息)
from kafka import KafkaConsumer
import timeconsumer = KafkaConsumer(bootstrap_servers=['172.21.10.136:9092'])
consumer.subscribe(topics=('test','test0'))
while True:msg = consumer.poll(timeout_ms=5)   #从kafka获取消息print msgtime.sleep(1)8、消费者(消息挂起与恢复)
from kafka import KafkaConsumer
from kafka.structs import TopicPartition
import timeconsumer = KafkaConsumer(bootstrap_servers=['172.21.10.136:9092'])
consumer.subscribe(topics=('test'))
consumer.topics()
consumer.pause(TopicPartition(topic=u'test', partition=0))
num = 0
while True:print numprint consumer.paused()   #获取当前挂起的消费者msg = consumer.poll(timeout_ms=5)print msgtime.sleep(2)num = num + 1if num == 10:print "resume..."consumer.resume(TopicPartition(topic=u'test', partition=0))print "resume......"pause执行后，consumer不能读取，直到调用resume后恢复。

示例代码2

原贴

"""生产者"""
def produce(self):producer = KafkaProducer(bootstrap_servers=self.bootstrap_servers)for i in range(4):msg = "msg%d" %iproducer.send(self.topic,key=str(i),value=msg)producer.close()"""一个消费者消费一个topic"""
def consume(self):#consumer = KafkaConsumer(self.topic,auto_offset_reset='earliest',group_id="testgroup",bootstrap_servers=self.bootstrap_servers)consumer = KafkaConsumer(self.topic,bootstrap_servers=self.bootstrap_servers)print consumer.partitions_for_topic(self.topic)  #获取test主题的分区信息
print consumer.topics()  #获取主题列表
print consumer.subscription()  #获取当前消费者订阅的主题
print consumer.assignment()  #获取当前消费者topic、分区信息
print consumer.beginning_offsets(consumer.assignment()) #获取当前消费者可消费的偏移量
consumer.seek(TopicPartition(topic=self.topic, partition=0), 1)  #重置偏移量，从第1个偏移量消费for message in consumer:print ("%s:%d:%d: key=%s value=%s" % (message.topic,message.partition,message.offset, message.key,message.value))"""一个消费者订阅多个topic """
def consume2(self):consumer = KafkaConsumer(bootstrap_servers=['192.168.124.201:9092'])
consumer.subscribe(topics=('TEST','TEST2'))  #订阅要消费的主题
print consumer.topics()
print consumer.position(TopicPartition(topic='TEST', partition=0)) #获取当前主题的最新偏移量
for message in consumer:print ("%s:%d:%d: key=%s value=%s" % (message.topic, message.partition,message.offset, message.key,message.value))"""消费者(手动拉取消息)"""
def consume3(self):consumer = KafkaConsumer(group_id="mygroup",max_poll_records=3,bootstrap_servers=['192.168.124.201:9092'])
consumer.subscribe(topics=('TEST','TEST2'))
while True:message = consumer.poll(timeout_ms=5)   #从kafka获取消息if message:print messagetime.sleep(1)

示例代码3

原贴

#coding=utf-8
from pykafka import KafkaClient
import codecs
import logging
logging.basicConfig(level = logging.INFO)client = KafkaClient(hosts = "172.16.82.163:9091")#生产kafka数据，通过字符串形式
def produce_kafka_data(kafka_topic):with kafka_topic.get_sync_producer() as producer:for i in range(4):producer.produce('test message' + str(i ** 2))#消费kafka数据
def consume_simple_kafka(kafka_topic, timeout):consumer = kafka_topic.get_simple_consumer(consumer_timeout_ms = timeout)for message in consumer:if message is not None:print message.offset, message.value#消费同一份kafka topic时，建议使用 get_balanced_consumer(),暂时不能使用
#问题：kazoo.handlers.threading.KazooTimeoutError: Connection time-out
def consume_kafka(kafka_topic, zkhost):balanced_consumer = kafka_topic.get_balanced_consumer(consumer_group = "testgroup",auto_commit_enable = False,zookeeper_connect = zkhost,#zookeeper = zkhost,zookeeper_connection_timeout_ms = 6000,consumer_timeout_ms = 10000,)for message in balanced_consumer:if message is not None:print message.offset, message.value#通过文件，往kafka刷数据
def produce_kafka_file(filename, kafka_topic):with kafka_topic.get_sync_producer() as producer:with codecs.open(filename, "r", "utf8") as rf:for line in rf:line = line.strip()if not line:continueproducer.produce(line)#===========================================================topic = client.topics["mytest"]#在consumer_timeout_ms内没有任何信息返回，则中断接受消息
cosumer = topic.get_simple_consumer(consumer_timeout_ms = 10000)
cnt = 0
for message in cosumer:if message is not None:print message.offset, message.valuecnt += 1
print cnt

kafka实战教程