如何从 python 客户端向 kafka 发送 JSON 对象

Question

我有一个简单的 JSON 对象，如下所示

d = { 'tag ': 'blah',
  'name' : 'sam',
  'score': 
    {'row1': 100,
      'row2': 200
     }
}

以下是我的 python 向 Kafka 发送消息的代码

from kafka import SimpleProducer, KafkaClient
import json 

# To send messages synchronously
kafka = KafkaClient('10.20.30.12:9092')
producer = SimpleProducer(kafka)
jd = json.dumps(d)
producer.send_messages(b'message1',jd)

我在风暴日志中看到正在接收消息，但它抛出元组转换 null { json 此处为结构 } 不确定需要做什么才能解决此问题？..

Answer 1

下面是我的生产者到kafka的代码。我唯一不同的是使用 yaml.safe_load 加载 json 内容。它 returns 内容作为字符串而不是 unicode。以下是摘录

with open('smaller_test_prod.txt') as f:
    for line in f:
        d = yaml.safe_load(line)
        jd = json.dumps(d)
        producer.send_messages(b'zeus_metrics',jd)

这里的每一行都是一个 json 数据存储在一个文件中。

Answer 2

Kafka 需要以字节为单位的值

b`some json message`

这是我的简单 Kafka 生产者，它将消息发送到 Kafka 服务器。

import json
from bson import json_util

from kafka import KafkaProducer

producer = KafkaProducer(bootstrap_servers='localhost:9092')

for i in range(10):
    data = { 'tag ': 'blah',
        'name' : 'sam',
        'index' : i,
        'score': 
            {'row1': 100,
             'row2': 200
        }
    }   
    producer.send('orders', json.dumps(data, default=json_util.default).encode('utf-8'))

此处json.dumps()将json转换为字符串，encode('utf-8')将字符串转换为字节数组。

如何从 python 客户端向 kafka 发送 JSON 对象

how to send JSON object to kafka from python client

python

json

apache-kafka

kafka-python