如何在 Kafka 中发布字典？答案

【问题标题】：How to publish dictionary in Kafka?如何在 Kafka 中发布字典？
【发布时间】：2019-04-07 21:02:20
【问题描述】：

我正在尝试在 kafka 主题中发布回复。此响应是从 mongodb 获取的。

from kafka import KafkaProducer
from kafka.errors import KafkaError
import json
import pymongo
from pymongo import MongoClient
import sys
import datetime

try:
    client = MongoClient('mongodb://A.B.C.D:27017/prod-production')
    db = client["prod-production"]
except Exception as e:
    print("Error occurred while connecting to DB")
    print(e)
producer = KafkaProducer(bootstrap_servers=['localhost:9092'])
producer = KafkaProducer(retries=5)
print("Initial time:")
print(datetime.datetime.now())
count = 1
for response in db.Response.find():
    if count >= 20:
        producer.flush()
        sys.exit()
    count += 1
    print(count)
    producer.send('example-topic', bytes(response))
print("Final time")
print(datetime.datetime.now())

我收到以下错误：

Traceback (most recent call last):   File "producer.py", line 28, in <module>
    producer.send('collect-production-response', bytes(response)) TypeError: 'str' object cannot be interpreted as an integer

但是，在python2中，不会出现这个错误。

【问题讨论】：

1) 使用 json.dumps 而不是 bytes 2) 我可能建议研究 Debezium 或 Kafka Connect 项目，以便让 Mongo 进入 Kafka
Python 2 将“str”和“bytes”视为同一事物——但将“unicode”视为不同的对象。 Python 3 将“str”和“unicode”视为同一事物 - 但将“bytes”视为不同的对象。
@cricket_007 请发布一个答案，这是您的评论，截至目前，以便我接受。

标签： python python-3.x apache-kafka

【解决方案1】：

我遇到了这个问题，我用以下两种方法解决了。

1- producer.send('example-topic', bytes(str(response), 'utf-8'))

或

2- producer.send('example-topic', str(response))

【讨论】：