我想给Kafka发一个大的csv。基本结构是读取csv的一行，并用标题压缩它。

a = dict(zip(header, line.split(",")

然后将其转换为json：

message = json.dumps(a)

然后我使用kafka python库发送消息

from kafka import SimpleProducer, KafkaClient
kafka = KafkaClient("localhost:9092")
producer = SimpleProducer(kafka)
producer.send_messages("topic", message)

使用pyspark，我已经很容易地从csv文件创建了一个消息的rdd

sc = SparkContext()
text = sc.textFile("file.csv")
header = text.first().split(',')
def remove_header(itr_index, itr):
    return iter(list(itr)[1:]) if itr_index == 0 else itr
noHeader = text.mapPartitionsWithIndex(remove_header)

messageRDD = noHeader.map(lambda x: json.dumps(dict(zip(header, x.split(","))

现在我想发送这些消息：我定义了一个函数

def sendkafka(message):
  kafka = KafkaClient("localhost:9092")
  producer = SimpleProducer(kafka)
  return producer.send_messages('topic',message)

然后我创建一个新的rdd来发送消息

sentRDD = messageRDD.map(lambda x: kafkasend(x))

然后我调用sentrdd.count（）
开始搅动和发送信息
不幸的是，这是非常缓慢的。它每秒发送1000条信息。这是一个10节点的集群，每个集群有4个CPU和8gb内存。
相比之下，在一个1000万行的csv上创建消息大约需要7秒~大约2gb
我认为问题是我在函数中示例化了一个kafka生产者。然而，如果我没有那么Spark抱怨生产者不存在，即使我已经尝试定义它全球。
也许有人能阐明如何解决这个问题。
谢谢您，