使用 python Spark 将大型 CSV 发送到 Kafka

Question

我正在尝试将大型 CSV 文件发送到 kafka。基本结构是读取一行 CSV 并将其压缩为 header.

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 会抱怨生产者不存在，即使我已经尝试在全球范围内定义它。

也许有人可以阐明如何解决这个问题。

谢谢，

Answer 1

您可以为每个分区创建一个生产者并使用 mapPartitions 或 foreachPartition:

def sendkafka(messages):
    kafka = KafkaClient("localhost:9092")
    producer = SimpleProducer(kafka)
    for message in messages:
        yield producer.send_messages('topic', message)

sentRDD = messageRDD.mapPartitions(sendkafka)

如果以上方法无法帮助您，您可以尝试使用 asynchronous producer.

来扩展它

在 Spark 2.x 中也可以使用 Kafka 数据源。您必须包含 spark-sql-kafka jar，匹配 Spark 和 Scala 版本（此处分别为 2.2.0 和 2.11）：

spark.jars.packages  org.apache.spark:spark-sql-kafka-0-10_2.11:2.2.0

将数据转换为 DataFrame（如果它还没有 DataFrame）：

messageDF = spark.createDataFrame(messageRDD, "string")

并使用 DataFrameWriter:

编写

(messageDF.write
    .format("kafka")
    .option("topic", topic_name)
    .option("kafka.bootstrap.servers", bootstrap_servers)
    .save())

使用 python Spark 将大型 CSV 发送到 Kafka

Sending Large CSV to Kafka using python Spark

python

apache-kafka

apache-spark

pyspark

kafka-python