深入了解如何基于Python读写Kafka

作者:Zl_one 时间:2021-02-13 09:33:09 

这篇文章主要介绍了深入了解如何基于Python读写Kafka,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下

本篇会给出如何使用python来读写kafka, 包含生产者和消费者.

以下使用kafka-python客户端

生产者

爬虫大多时候作为消息的发送端, 在消息发出去后最好能记录消息被发送到了哪个分区, offset是多少, 这些记录在很多情况下可以帮助快速定位问题, 所以需要在send方法后加入callback函数, 包括成功和失败的处理


# -*- coding: utf-8 -*-

'''
callback也是保证分区有序的, 比如2条消息, a先发送, b后发送, 对于同一个分区, 那么会先回调a的callback, 再回调b的callback
'''

import json
from kafka import KafkaProducer

topic = 'demo'

def on_send_success(record_metadata):
 print(record_metadata.topic)
 print(record_metadata.partition)
 print(record_metadata.offset)

def on_send_error(excp):
 print('I am an errback: {}'.format(excp))

def main():
 producer = KafkaProducer(
   bootstrap_servers='localhost:9092'
 )
 producer.send(topic, value=b'{"test_msg":"hello world"}').add_callback(on_send_success).add_callback(
   on_send_error)
 # close() 方法会阻塞等待之前所有的发送请求完成后再关闭 KafkaProducer
 producer.close()

def main2():
 '''
 发送json格式消息
 :return:
 '''
 producer = KafkaProducer(
   bootstrap_servers='localhost:9092',
   value_serializer=lambda m: json.dumps(m).encode('utf-8')
 )
 producer.send(topic, value={"test_msg": "hello world"}).add_callback(on_send_success).add_callback(
   on_send_error)
 # close() 方法会阻塞等待之前所有的发送请求完成后再关闭 KafkaProducer
 producer.close()
if __name__ == '__main__':
 # main()
 main2()

消费者

kafka的消费模型比较复杂, 我会分以下几种情况来进行说明

1.不使用消费组(group_id=None)

不使用消费组的情况下可以启动很多个消费者, 不再受限于分区数, 即使消费者数量 > 分区数, 每个消费者也都可以收到消息


# -*- coding: utf-8 -*-

'''
消费者: group_id=None
'''
from kafka import KafkaConsumer
topic = 'demo'
def main():
 consumer = KafkaConsumer(
   topic,
   bootstrap_servers='localhost:9092',
   auto_offset_reset='latest',
   # auto_offset_reset='earliest',
 )
 for msg in consumer:
   print(msg)
   print(msg.value)
 consumer.close()
if __name__ == '__main__':
 main()

2.指定消费组

以下使用pool方法来拉取消息

pool 每次拉取只能拉取一个分区的消息, 比如有2个分区1个consumer, 那么会拉取2次

pool 是如果有消息马上进行拉取, 如果timeout_ms内没有新消息则返回空dict, 所以可能出现某次拉取了1条消息, 某次拉取了max_records条


# -*- coding: utf-8 -*-

'''
消费者: 指定group_id
'''

from kafka import KafkaConsumer

topic = 'demo'
group_id = 'test_id'

def main():
 consumer = KafkaConsumer(
   topic,
   bootstrap_servers='localhost:9092',
   auto_offset_reset='latest',
   group_id=group_id,

)
 while True:
   try:
     # return a dict
     batch_msgs = consumer.poll(timeout_ms=1000, max_records=2)
     if not batch_msgs:
       continue
     '''
     {TopicPartition(topic='demo', partition=0): [ConsumerRecord(topic='demo', partition=0, offset=42, timestamp=1576425111411, timestamp_type=0, key=None, value=b'74', headers=[], checksum=None, serialized_key_size=-1, serialized_value_size=2, serialized_header_size=-1)]}
     '''
     for tp, msgs in batch_msgs.items():
       print('topic: {}, partition: {} receive length: '.format(tp.topic, tp.partition, len(msgs)))
       for msg in msgs:
         print(msg.value)
   except KeyboardInterrupt:
     break

consumer.close()

if __name__ == '__main__':
 main()

关于消费组

我们根据配置参数分为以下几种情况

  • group_id=None

    • auto_offset_reset='latest': 每次启动都会从最新出开始消费, 重启后会丢失重启过程中的数据

    • auto_offset_reset='latest': 每次从最新的开始消费, 不会管哪些任务还没有消费

  • 指定group_id

    • auto_offset_reset='latest': 从上次提交offset的地方开始消费

    • auto_offset_reset='earliest': 从上次提交offset的地方开始消费

    • auto_offset_reset='latest': 只消费启动后的收到的数据, 重启后会从上次提交offset的地方开始消费

    • auto_offset_reset='earliest': 从最开始消费全量数据

    • 全新group_id

    • 旧group_id(即kafka集群中还保留着该group_id的提交记录)

    性能测试

    以下是在本地进行的测试, 如果要在线上使用kakfa, 建议提前进行性能测试

    producer


    # -*- coding: utf-8 -*-

    '''
    producer performance

    environment:
     mac
     python3.7
     broker 1
     partition 2
    '''

    import json
    import time
    from kafka import KafkaProducer

    topic = 'demo'
    nums = 1000000

    def main():
     producer = KafkaProducer(
       bootstrap_servers='localhost:9092',
       value_serializer=lambda m: json.dumps(m).encode('utf-8')
     )
     st = time.time()
     cnt = 0
     for _ in range(nums):
       producer.send(topic, value=_)
       cnt += 1
       if cnt % 10000 == 0:
         print(cnt)

    producer.flush()

    et = time.time()
     cost_time = et - st
     print('send nums: {}, cost time: {}, rate: {}/s'.format(nums, cost_time, nums // cost_time))

    if __name__ == '__main__':
     main()

    '''
    send nums: 1000000, cost time: 61.89236712455749, rate: 16157.0/s
    send nums: 1000000, cost time: 61.29534196853638, rate: 16314.0/s
    '''

    consumer


    # -*- coding: utf-8 -*-

    '''
    consumer performance
    '''

    import time
    from kafka import KafkaConsumer

    topic = 'demo'
    group_id = 'test_id'

    def main1():
     nums = 0
     st = time.time()

    consumer = KafkaConsumer(
       topic,
       bootstrap_servers='localhost:9092',
       auto_offset_reset='latest',
       group_id=group_id
     )
     for msg in consumer:
       nums += 1
       if nums >= 500000:
         break
     consumer.close()

    et = time.time()
     cost_time = et - st
     print('one_by_one: consume nums: {}, cost time: {}, rate: {}/s'.format(nums, cost_time, nums // cost_time))

    def main2():
     nums = 0
     st = time.time()

    consumer = KafkaConsumer(
       topic,
       bootstrap_servers='localhost:9092',
       auto_offset_reset='latest',
       group_id=group_id
     )
     running = True
     batch_pool_nums = 1
     while running:
       batch_msgs = consumer.poll(timeout_ms=1000, max_records=batch_pool_nums)
       if not batch_msgs:
         continue
       for tp, msgs in batch_msgs.items():
         nums += len(msgs)
         if nums >= 500000:
           running = False
           break

    consumer.close()

    et = time.time()
     cost_time = et - st
     print('batch_pool: max_records: {} consume nums: {}, cost time: {}, rate: {}/s'.format(batch_pool_nums, nums,
                                                 cost_time,
                                                 nums // cost_time))

    if __name__ == '__main__':
     # main1()
     main2()

    '''
    one_by_one: consume nums: 500000, cost time: 8.018627166748047, rate: 62354.0/s
    one_by_one: consume nums: 500000, cost time: 7.698841094970703, rate: 64944.0/s

    batch_pool: max_records: 1 consume nums: 500000, cost time: 17.975456953048706, rate: 27815.0/s
    batch_pool: max_records: 1 consume nums: 500000, cost time: 16.711708784103394, rate: 29919.0/s

    batch_pool: max_records: 500 consume nums: 500369, cost time: 6.654940843582153, rate: 75187.0/s
    batch_pool: max_records: 500 consume nums: 500183, cost time: 6.854053258895874, rate: 72976.0/s

    batch_pool: max_records: 1000 consume nums: 500485, cost time: 6.504687070846558, rate: 76942.0/s
    batch_pool: max_records: 1000 consume nums: 500775, cost time: 7.047331809997559, rate: 71058.0/s
    '''

    来源:https://www.cnblogs.com/zlone/p/12116817.html

    标签:Python,Kafka,读写
    0
    投稿

    猜你喜欢

  • PHP基于迭代实现文件夹复制、删除、查看大小等操作的方法

    2023-09-03 22:21:42
  • python制作抽奖程序代码详解

    2022-05-04 08:51:18
  • 小议sqlserver数据库主键选取策略

    2024-01-25 15:03:52
  • python深度学习tensorflow入门基础教程示例

    2021-08-25 07:03:08
  • 一文带你搞懂JS中导入模块import和require的区别

    2023-07-21 03:24:18
  • Django框架静态文件使用/中间件/禁用ip功能实例详解

    2023-06-12 02:48:38
  • Mysql数据库常用命令

    2009-03-06 14:29:00
  • Bootstrap导航条学习使用(二)

    2024-05-02 17:31:16
  • PHP OPP机制和模式简介(抽象类、接口和契约式编程)

    2023-11-21 10:53:53
  • SQLServer 设置单词首字母大写

    2024-01-12 20:14:41
  • python docx的超链接网址和链接文本操作

    2021-06-05 22:10:55
  • 解决Python报错问题[SSL: SSLV3_ALERT_HANDSHAKE_FAILURE]

    2021-08-20 23:26:07
  • Python通过30秒就能学会的漂亮短程序代码(过程全解)

    2022-08-22 13:54:50
  • PHP操作数组的一些函数整理介绍

    2023-11-24 14:24:17
  • 通过 Django Pagination 实现简单分页功能

    2021-03-15 15:53:03
  • python检测某个变量是否有定义的方法

    2021-05-17 18:45:01
  • python 匹配url中是否存在IP地址的方法

    2023-04-13 14:29:08
  • SQL Server数据库的高性能优化经验总结

    2024-01-18 17:04:55
  • python缺失值填充方法示例代码

    2023-06-28 12:23:24
  • Python匿名函数/排序函数/过滤函数/映射函数/递归/二分法

    2022-03-10 21:04:56
  • asp之家 网络编程 m.aspxhome.com