ConFluent Kafka Python库为批量msg配置生产者

Question

我需要设置Kafka生产商发送500 msg在一批不是由味精味精，但批量进口味精。我查过了https://github.com/dpkp/kafka-python/issues/479并尝试了制作人。发送消息（主题，*消息）但失败并出现错误：

>       producer.send_messages('event_connector_mt', *load_entries)
E       AttributeError: 'cimpl.Producer' object has no attribute 'send_messages'

我也试着像《代码》制作人一样通过考试。制作（主题，*消息）失败：

       producer.produce('event_connector_mt', *load_entries)
E       TypeError: function takes at most 8 arguments (501 given)

因此，我挖掘了更多信息，发现我必须在producer配置中将类型设置为async和batch。大小要大于默认值，但当我尝试配置时，如：

from confluent_kafka import Consumer, Producer       

producer = Producer(**{'bootstrap.servers': KAFKA_BROKERS,
                       'queue.buffering.max.messages': 1000000,
                       'batch.num.messages': 500,
                       'batch.size': 19999,
                       'producer.type': 'async'
                       })

失败：

E       KafkaException: KafkaError{code=_INVALID_ARG,val=-186,str="No such configuration property: "producer.type""}

批处理也有同样的错误。你能告诉我在哪里以及如何设置异步和批量大小，或者以任何其他方式将批量MSG传递给Kafka 0.9.3.1吗

Answer 1

你正在混淆名为“汇合-Kafka-蟒蛇”的汇合蟒蛇客户端

http://docs.confluent.io/3.2.1/clients/confluent-kafka-python/

使用“Kafkapython”客户端

https://github.com/dpkp/kafka-python

这是两个具有不同API的不同客户端。

Answer 2

默认情况下，所有生成器都是异步的。Producer.type和batch.size不支持底层库的配置。

因此，请使用可用配置批处理。num.messages或message。最大字节数。