时间戳

首页 > 解决方案 > 通过 KAFKA 发送带有多个 JSON 对象的 JSON 文件

python - 通过 KAFKA 发送带有多个 JSON 对象的 JSON 文件

问题描述

我有一个文件，其中包含以下格式的多个 json 文档。

{"attribute1": "value1", "attribute2": "value2", "attribute3": "value3", "attribute4": "value4"} {"attribute1": "value11", "attribute2": "value12", "attribute3": "value13", "attribute4": "value14"} {"attribute1": "value21", "attribute22": "value2", "attribute23": "value3", "attribute4": "value24"}

我正在尝试将单个 json 文档发送到 kafka。该脚本以退出代码 0 执行，但我看不到 KAFKA 消费者上没有任何消息。我不确定我哪里出错了。

我的代码如下：

import csv
import json

bootstrap = ['hostname:9092']
valueSerializer = lambda x: dumps(x).encode('utf-8')

producer = KafkaProducer(bootstrap_servers = bootstrap, value_serializer = valueSerializer)

table = []
with open('~/json_file_name.json', 'r') as json_file:
    for line in json_file:
        table.append(json.loads(line))

#numrows = len(table)
#print(numrows)

for row in table:
    print(row)
    producer.send('Topic_Name', value=row)

标签： pythonjsonpython-3.xapache-kafkakafka-python

解决方案

您可能没有为生产者发送足够的数据来刷新其批次。您尚未显示 KafkaProducer 的导入，但请查看是否可以producer.flush()在脚本末尾执行

顺便说一句，您不需要表变量，只需在读取文件行时发送即可。您也不需要，dumps(x)因为您正在发送json.loads已经获得的字符串

您还可以删除 csv 导入

推荐阅读