通过python操作kafka

简介:

通过python操作kafka


kafka特性:

通过O(1)的磁盘数据结构提供消息的持久化,这种结构对于即使数以TB的消息存储也能够保持长时间的稳定性能。

高吞吐量[2] :即使是非常普通的硬件Kafka也可以支持每秒数百万[2] 的消息

支持通过Kafka服务器和消费机集群来分区消息

支持Hadoop并行数据加载


术语:

Broker

Kafka集群包含一个或多个服务器,这种服务器被称为broker

Topic

每条发布到Kafka集群的消息都有一个类别,这个类别被称为Topic。(物理上不同Topic的消息分开存储,逻辑上一个Topic的消息虽然保存于一个或多个broker上但用户只需指定消息的Topic即可生产或消费数据而不必关心数据存于何处)

Partition

Partition是物理上的概念,每个Topic包含一个或多个Partition.

Producer

负责发布消息到Kafka broker

Consumer

消息消费者,向Kafka broker读取消息的客户端。

Consumer Group

每个Consumer属于一个特定的Consumer Group(可为每个Consumer指定group name,若不指定group name则属于默认的group)。

一、安装

在pypi.python.org有很多关于操作kafka的组件,我们选择weight最高的kafka 1.3.5

1、有internet网的情况下执行如下命令安装:


pip install kafka
easy_install kafka

2、无internet网的情况下把源码下载下来,上传到需要安装的主机

压缩包:kafka-x.x.x.tar.gz

解压: tar xvf kafka-x.x.x.tar.gz

执行安装命令: cd kafka-x.x.x

python setup.py install

如安装报依赖错误,需要把依赖的组件也下载下来,然后进行安装,同样的方法,不赘述!

二、按照官网的样例,先跑一个应用

1、生产者:

from kafka import KafkaProducer

producer = KafkaProducer(bootstrap_servers=['xxx.xx.xx.xxx: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
producer.send('test', msg)
producer.close()

2、消费者(简单demo):

from kafka import KafkaConsumer

consumer = KafkaConsumer('test',
bootstrap_servers=['xxx.xx.xx.xx: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 KafkaConsumer

consumer = KafkaConsumer('test',
group_id='my-group',
bootstrap_servers=['xxx.xx.xx.xx: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 KafkaConsumer

consumer = KafkaConsumer('test',
auto_offset_reset='earliest',
bootstrap_servers=['xxx.xx.xx.xxx: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 TopicPartition

consumer = KafkaConsumer('test',
bootstrap_servers=['xxx.xx.xx.xxx: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 TopicPartition

consumer = KafkaConsumer(bootstrap_servers=['xxx.xx.xx.xxx: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 time

consumer = KafkaConsumer(bootstrap_servers=['xxx.xx.xx.xxx:9092'])
consumer.subscribe(topics=('test','test0'))
while True:
msg = consumer.poll(timeout_ms=5) #从kafka获取消息
print msg
time.sleep(1)

8、消费者(消息挂起与恢复)

from kafka import KafkaConsumer
from kafka.structs import TopicPartition
import time

consumer = KafkaConsumer(bootstrap_servers=['xxx.xx.xx.xxx:9092'])
consumer.subscribe(topics=('test'))
consumer.topics()
consumer.pause(TopicPartition(topic=u'test', partition=0))
num = 0
while True:
print num
print consumer.paused() #获取当前挂起的消费者
msg = consumer.poll(timeout_ms=5)
print msg
time.sleep(2)
num = num + 1
if num == 10:
print "resume..."
consumer.resume(TopicPartition(topic=u'test', partition=0))
print "resume......"
pause执行后,consumer不能读取,直到调用resume后恢复。


相关文章
|
2月前
|
消息中间件 Kafka Python
|
2月前
|
消息中间件 Kafka API
python之kafka日志
python之kafka日志
25 3
|
4月前
|
存储 监控 数据处理
💻Python高手必备!文件系统操作秘籍,让你的数据存取如臂使指
【7月更文挑战第29天】在数据驱动时代, Python以简洁语法、丰富库生态和强大跨平台能力, 成为数据科学等领域首选。本文探讨Python文件系统操作秘籍, 助力高效数据处理。
47 11
|
4月前
|
安全 数据安全/隐私保护 Python
|
4月前
|
监控 网络协议 网络安全
SMTP操作使用详解并通过python进行smtp邮件发送示例
SMTP操作使用详解并通过python进行smtp邮件发送示例
114 3
|
4月前
|
数据挖掘 数据处理 Python
🔍深入Python系统编程腹地:文件系统操作与I/O管理,打造高效数据处理流水线
【7月更文挑战第29天】深入Python系统编程腹地:文件系统操作与I/O管理,打造高效数据处理流水线
34 3
|
4月前
|
索引 Python
Python的列表操作有哪些?
Python的列表操作非常丰富,包括列表的创建、元素的访问、修改、添加、删除、切片、排序等多个方面。
41 12
|
4月前
|
API Python
Python高手修炼手册:精通文件系统操作,掌控I/O管理,提升编程效率
【7月更文挑战第30天】在 Python 编程中, 文件系统操作与 I/O 管理是连接程序与数据的关键。初学者常因路径错误和权限问题受挫, 而高手能自如管理文件。传统 `os` 和 `os.path` 模块易出错, `pathlib` 提供了更直观的对象导向 API。I/O 方面, 同步操作会阻塞程序, 异步 (如使用 `aiofiles`) 则能大幅提升并发能力。真正的高手不仅掌握 API, 更能预见性能瓶颈并优化代码, 实现高效与优雅。
38 1
|
4月前
|
Serverless 语音技术 开发工具
函数计算操作报错合集之怎么何集成nls tts python sdk
在使用函数计算服务(如阿里云函数计算)时,用户可能会遇到多种错误场景。以下是一些常见的操作报错及其可能的原因和解决方法,包括但不限于:1. 函数部署失败、2. 函数执行超时、3. 资源不足错误、4. 权限与访问错误、5. 依赖问题、6. 网络配置错误、7. 触发器配置错误、8. 日志与监控问题。