[Kafka] Python에서 AssertionError 발생

이전 포스팅에서 kafka 패키지를 kafka-python으로 변경을 하니 이제는 다른 에러가 발생하게 되었다. 

 

실행코드

from kafka import KafkaProducer

producer = KafkaProducer(bootstrap_servers='localhost:9092')
producer.send('test', 'Hello World!!!!')

 

코드를 보면 별다른 한글도 없는 전혀 문제가 없을 법한 코드인데 아래와 같은 에러가 발생하였다.

C:\Users\user\anaconda3\python.exe D:/phantom/source-python/main.py
Traceback (most recent call last):
  File "D:/phantom/source-python/main.py", line 4, in <module>
    producer.send('test', 'Hello World')
  File "C:\Users\user\anaconda3\lib\site-packages\kafka\producer\kafka.py", line 585, in send
    assert type(value_bytes) in (bytes, bytearray, memoryview, type(None))
AssertionError

Process finished with exit code 1

 

문제해결

버전 이슈라서 encode를 하라는 내용

 

from kafka import KafkaProducer

producer = KafkaProducer(bootstrap_servers='localhost:9092')
producer.send('test', 'Hello World!!!'.encode('utf-8'))

문자열에 .encode('utf-8')을 추가하니 에러 없이 코드가 실행되었다. 그러나 전반적으로 위의 코드들은 올드한 것으로 보이며, 아래와 같이 테스트 코드를 만들어서 처리를 하였음

 

from kafka import KafkaProducer
from time import sleep
from json import dumps

producer = KafkaProducer(acks=1, compression_type='gzip', bootstrap_servers=['localhost:9092'],
                         value_serializer=lambda x: dumps(x).encode('utf-8'))

for i in range(100):
    data = {'str' : 'result'+str(i)}
    producer.send('test', value=data)
    sleep(5)

 

 

참고자료

https://github.com/scrapinghub/frontera/issues/297