python向kafka发送json数据
1.安装库2.创建python文件3.使用OffsetExploret查看kafka收到的该消息
·
1.安装kafka支持库
pip install kafka-python
2.创建python文件
参数key_serializer用来指定键的序列化的方式,参数value_serializer用来指定值的序列化的方式,这里使用 json 来序列化数据,从而实现向 Kafka 传入一个字典,Kafka 自动把它转成 JSON 字符串的效果。
from kafka import KafkaProducer
from kafka.errors import kafka_errors
import traceback
import json
producer = KafkaProducer(
bootstrap_servers=['192.168.1.214:9092'], # kafka服务器ip和端口号
key_serializer=lambda k: json.dumps(k).encode(), # 假设生产的消息为json字符串
value_serializer=lambda v: json.dumps(v).encode()) # 假设生产的消息为json字符串
data = {'msg': 'hello kafka!'}
future = producer.send(
'FaultRecordLog', # 要发送的kafka主题
key='mytopic', # 同一个key值,会被送至同一个分区
value=data,
partition=0) # 向分区1发送消息
print("send {}".format(str(data)))
try:
future.get(timeout=10) # 监控是否发送成功
except kafka_errors: # 发送失败抛出kafka_errors
traceback.format_exc()
3.运行该python文件
4.使用OffsetExplorer 查看kafka收到的该消息
下一篇文章是消费数据python消费kafka数据_grfstc的博客-CSDN博客
更多推荐
已为社区贡献9条内容
所有评论(0)