我有一个大的csv，我想写一个Kafka主题。
def producer():
producer = KafkaProducer(bootstrap_servers='mykafka-broker')
with open('/home/antonis/repos/testfile.csv') as file:
reader = csv.DictReader(file, delimiter=";")
for row in reader:
producer.send(topic='stable_topic', value=row)
producer.flush()
if __name__ == '__main__':
producer()此代码产生错误：
AssertionError: value must be bytes文件如下所示：
"timestamp","name","age"
2020-03-01 00:00:01,John,36
2020-03-01 00:00:01,Peter,22有人能帮我吗？
2条答案
按热度按时间tnkciper1#
您需要正确地序列化您的值。
下面应该可以做到这一点：
4ngedf3f2#
与其重新发明轮子，不如使用已经存在的非常好的轮子：）它是kafka connect，它是apache kafka的一部分。
有几个连接器可以从csv读取，包括kafka connect spooldir（参见示例）和filepulse。
了解更多关于Kafka连接在这个演讲。