使用Python，如何产生KAFKA消息超过1MB？

Question

我试图通过Kafka传输一些巨大的JSON数据，但没有发生。 Python生产商也没有任何错误，也没有消费者端的任何消息。

producer = KafkaProducer(bootstrap_servers="kafka_server")
product_message = json.dumps(data)
producer.send(kafka_topic, key=bytes('kk', 'utf-8'), value=bytes(product_message, 'utf-8'))

Answer 1

您可能需要增加主题配置 > 能够产生该消息。

另外，您可能可以通过获得发送返回的未来来了解发生的事情： https://stackover.com/a> https：/ /55538034/19059974

Answer 2

我有相同的错误，并且通过阅读这个线程我已经解决了这样的问题：

假设您有文件server.properties和computer.properties在路径/usr/usr/local/kafka/kafka/config/ （可以在您的环境变量中定义此路径为$ kafka_home，但不必这样做）

并且您想将最大消息长度增加到15728640（15 MB） ），

服务器端（= broker）

sudo vim /usr/local/kafka/config/server.properties

添加行或更改变量值

# my additions
# increase max message size to 15 MB
message.max.bytes=15728640 
replica.fetch.max.bytes=15728640

消费者端

sudo vim /usr/local/kafka/config/consumer.properties

添加行或更改变量值

# my additions
# increase max message size to 15 MB
message.max.bytes=15728640
max.partition.fetch.bytes=15728640
fetch.max.bytes=15728640

重新启动服务

使更改有效（手动运行这些命令）

sudo systemctl daemon-reload  # wait a little

sudo systemctl restart zookeeper  # wait a little

sudo systemctl restart kafka  # wait a little

确保它们通过 如果没有运行

sudo systemctl status zookeeper
sudo systemctl status kafka

，请再次重新启动

生产者方面，在Python模块Kafka-Python的背景下

，您必须添加K-Argument max_request_size（

producer.send(kafka_topic, key=bytes('kk', 'utf-8'), value=bytes(product_message, 'utf-8'))

用行替换行

producer.send(kafka_topic, key=bytes('kk', 'utf-8'), value=bytes(product_message, 'utf-8'), max_request_size=15728640)