Ошибка PYspark при отправке RDD в Kafka Тема

Question

HI все, что я немного смущен этим.

мой искровой код генерирует ожидаемый результат, однако, как только я попытаюсь отправить RDD обратно в kafka, это ошибка каждый раз. Я не совсем уверен, что не так с моим кодом.

def sendkafka(messages): 
    kafka = KafkaClient(kafkaip) 
    producer = SimpleProducer(kafka, 'dev_test_upstream') 
    for message in messages: 
     producer.send_messages(message) 

Answer 1

Вы уверены, что твой метод sendkafka() работает должным образом. Похоже, вы предоставили частичный листинг, поэтому не уверены, что эта часть работает. У меня есть пример того, как отправлять данные в Kafka из Python, посмотрите Python client for publishing and consuming message from Apache Kafka, вы можете использовать метод manufacturer.py вместо метода sendKafka()