как отправить объект JSON Кафки шторм из питона клиента

Question

У меня есть простой объект JSON как следующий

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

Ниже мой питон код, который отправляет сообщения Кафки

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) 

I см. в журналах шторма, что сообщение получено, но его метание Ошибка преобразования для tuple {json structure здесь} не уверены, что нужно сделать, чтобы исправить это? ..

Answer 1

Ниже представлен мой код для продюсера кафке. Единственное, что я сделал по-другому, это использовать yaml.safe_load для загрузки содержимого json. Он возвращает содержимое в виде строк вместо 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

Кафка ожидает значения в байтах

b`some json message` 

Вот мой простой продюсер Кафка, который отправляет сообщение на сервер Кафки.

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(d, indent=4, default=json_util.default).encode('utf-8')) 

Здесь json.dumps() преобразовать в строку JSON и кодирования ('UTF-8') преобразует строку в массив байтов.