将RDD发送到卡夫卡时发生PYspark错误主题

Question

您好，我对此有点困惑。

我的火花代码会生成预期的输出，但是一旦我尝试将RDD发送回卡夫卡，它就会每次发生错误。我不太清楚我的代码有什么问题。

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

Answer 1

您确定您的sendkafka（）方法按预期工作。它看起来像你提供了部分清单，所以不知道该部分是否工作。我有一个如何从Python发送数据给Kafka的例子，看看Python client for publishing and consuming message from Apache Kafka，你可能想使用producer.py而不是sendKafka（）方法