python 消费 kafka 数据教程
1.安装python模块
pip install --user kafka-python==1.4.3
如果报错压缩相关的错尝试安装下面的依赖
yum install snappy-devel yum install lz4-devel pip install python-snappy pip install lz4
2.生产者
#!/usr/bin/env python # coding : utf-8 from kafka import KafkaProducer import json def kafkaProducer(): producer = KafkaProducer(bootstrap_servers='ip:9092',value_serializer=lambda v: json.dumps(v).encode('utf-8')) producer.send('world', {'key1': 'value1'}) if __name__ == '__main__': kafkaProducer()
2.消费者
from kafka import KafkaConsumer from kafka.structs import TopicPartition import time import click import ConfigParser import json import threading import datetime import sched config = ConfigParser.ConfigParser() config.read("amon.ini") @click.group() def cli(): pass @cli.command() @click.option('--topic',type=str) @click.option('--offset', type=click.Choice(['smallest', 'earliest', 'largest'])) @click.option("--group",type=str) def client(topic,offset,group): click.echo(topic) consumer = KafkaConsumer(topic, bootstrap_servers=config.get("KAFKA", "Broker_Servers").split(","), group_id=group, auto_offset_reset=offset) for message in consumer: click.echo(message.value) # click.echo("%d:%d: key=%s value=%s" % (message.partition, # message.offset, message.key, # message.value)) if __name__ == '__main__': cli()
3.多线程消费
#coding:utf-8 import threading import os import sys from kafka import KafkaConsumer, TopicPartition, OffsetAndMetadata from collections import OrderedDict threads = [] class MyThread(threading.Thread): def __init__(self, thread_name, topic, partition): threading.Thread.__init__(self) self.thread_name = thread_name self.partition = partition self.topic = topic def run(self): print("Starting " + self.name) Consumer(self.thread_name, self.topic, self.partition) def stop(self): sys.exit() def Consumer(thread_name, topic, partition): broker_list = 'ip1:9092,ip2:9092' ''' fetch_min_bytes(int) - 服务器为获取请求而返回的最小数据量,否则请等待 fetch_max_wait_ms(int) - 如果没有足够的数据立即满足fetch_min_bytes给出的要求,服务器在回应提取请求之前将阻塞的最大时间量(以毫秒为单位) fetch_max_bytes(int) - 服务器应为获取请求返回的最大数据量。这不是绝对最大值,如果获取的第一个非空分区中的第一条消息大于此值, 则仍将返回消息以确保消费者可以取得进展。注意:使用者并行执行对多个代理的提取,因此内存使用将取决于包含该主题分区的代理的数量。 支持的Kafka版本> = 0.10.1.0。默认值:52428800(50 MB)。 enable_auto_commit(bool) - 如果为True,则消费者的偏移量将在后台定期提交。默认值:True。 max_poll_records(int) - 单次调用中返回的最大记录数poll()。默认值:500 max_poll_interval_ms(int) - poll()使用使用者组管理时的调用之间的最大延迟 。这为消费者在获取更多记录之前可以闲置的时间量设置了上限。 如果 poll()在此超时到期之前未调用,则认为使用者失败,并且该组将重新平衡以便将分区重新分配给另一个成员。默认300000 ''' consumer = KafkaConsumer(bootstrap_servers=broker_list, group_id="test000001", client_id=thread_name, enable_auto_commit=False, fetch_min_bytes=1024 * 1024, # 1M # fetch_max_bytes=1024 * 1024 * 1024 * 10, fetch_max_wait_ms=60000, # 30s request_timeout_ms=305000, # consumer_timeout_ms=1, # max_poll_records=5000, ) # 设置topic partition tp = TopicPartition(topic, partition) # 分配该消费者的TopicPartition,也就是topic和partition,根据参数,每个线程消费者消费一个分区 consumer.assign([tp]) #获取上次消费的最大偏移量 offset = consumer.end_offsets([tp])[tp] print(thread_name, tp, offset) # 设置消费的偏移量 consumer.seek(tp, offset) print u"程序首次运行\t线程:", thread_name, u"分区:", partition, u"偏移量:", offset, u"\t开始消费..." num = 0 # 记录该消费者消费次数 while True: msg = consumer.poll(timeout_ms=60000) end_offset = consumer.end_offsets([tp])[tp] '''可以自己记录控制消费''' print u'已保存的偏移量', consumer.committed(tp), u'最新偏移量,', end_offset if len(msg) > 0: print u"线程:", thread_name, u"分区:", partition, u"最大偏移量:", end_offset, u"有无数据,", len(msg) lines = 0 for data in msg.values(): for line in data: print line lines += 1 ''' do something ''' # 线程此批次消息条数 print(thread_name, "lines", lines) if True: # 可以自己保存在各topic, partition的偏移量 # 手动提交偏移量 offsets格式:{TopicPartition:OffsetAndMetadata(offset_num,None)} consumer.commit(offsets={tp: (OffsetAndMetadata(end_offset, None))}) if True == 0: # 系统退出?这个还没试 os.exit() ''' sys.exit() 只能退出该线程,也就是说其它两个线程正常运行,主程序不退出 ''' else: os.exit() else: print thread_name, '没有数据' num += 1 print thread_name, "第", num, "次" if __name__ == '__main__': try: t1 = MyThread("Thread-0", "test", 0) threads.append(t1) t2 = MyThread("Thread-1", "test", 1) threads.append(t2) t3 = MyThread("Thread-2", "test", 2) threads.append(t3) for t in threads: t.start() for t in threads: t.join() print("exit program with 0") except: print("Error: failed to run consumer program")
参考:https://kafka-python.readthedocs.io/en/master/index.html
https://www.jb51.net/article/176911.htm
以上这篇python 消费 kafka 数据教程就是小编分享给大家的全部内容了,希望能给大家一个参考,也希望大家多多支持亿速云。