对python操作kafka写入json数据的简单demo分享

yipeiwu_com6年前Python基础

如下所示:

安装kafka支持库pip install kafka-python
from kafka import KafkaProducer
import json
 
'''
 生产者demo
 向test_lyl2主题中循环写入10条json数据
 注意事项:要写入json数据需加上value_serializer参数,如下代码
'''
producer = KafkaProducer(
       value_serializer=lambda v: json.dumps(v).encode('utf-8'),
       bootstrap_servers=['192.168.12.101:6667','192.168.12.102:6667','192.168.12.103:6667']
       )
for i in range(10):
 data={
  "name":"李四",
  "age":23,
  "gender":"男",
  "id":i
 }
 producer.send('test_lyl2', data)
producer.close()
from kafka import KafkaConsumer
import json
 
'''
 消费者demo
 消费test_lyl2主题中的数据
 注意事项:如需以json格式读取数据需加上value_deserializer参数
'''
 
 
consumer = KafkaConsumer('test_lyl2',group_id="lyl-gid1",
       bootstrap_servers=['192.168.12.101:6667','192.168.12.102:6667','192.168.12.103:6667'],
       auto_offset_reset='earliest',value_deserializer=json.loads
       )
for message in consumer:
 print(message.value)

以上这篇对python操作kafka写入json数据的简单demo分享就是小编分享给大家的全部内容了,希望能给大家一个参考,也希望大家多多支持【听图阁-专注于Python设计】。

相关文章

Django model select的多种用法详解

Django model select的多种用法详解

《Django model update的各种用法介绍》文章介绍了Django model的各种update操作,这篇文章就是她的姊妹篇,详细介绍Django model select的...

对Python中DataFrame按照行遍历的方法

在做分类模型时候,需要在DataFrame中按照行获取数据以便于进行训练和测试。 import pandas as pd dict=[[1,2,3,4,5,6],[2,3,4,5,6...

python pillow模块使用方法详解

pillow Pillow是PIL的一个派生分支,但如今已经发展成为比PIL本身更具活力的图像处理库。pillow可以说已经取代了PIL,将其封装成python的库(pip即可安装),...

Python cookbook(数据结构与算法)将名称映射到序列元素中的方法

本文实例讲述了Python将名称映射到序列元素中的方法。分享给大家供大家参考,具体如下: 问题:希望通过名称来访问元素,减少结构中对位置的依赖性 解决方案:使用命名元组collectio...

使用Python获取Linux系统的各种信息

在本文中,我们将会探索使用Python编程语言工具来检索Linux系统各种信息。走你。 哪个Python版本? 当我提及Python,所指的就是CPython 2(准确的是2.7...