将数据帧写入kafka pyspark

Question

我有一个 spark 数据框，我想将其写入 Kafka。我在sn-p下面试过了，

from kafka import KafkaProducer
producer = KafkaProducer(bootstrap_servers = util.get_broker_metadata())
df = sqlContext.createDataFrame([("foo", 1), ("bar", 2), ("baz", 3)], ('k', 'v'))
for row in df.rdd.collect():
    producer.send('topic',str(row.asDict()))
    producer.flush()

这可行，但这个 sn-p 的问题是它不是可扩展的，因为每次收集运行时，数据都会在驱动程序节点上聚合，并且会减慢所有操作。

因为数据帧上的 foreach 操作可以在工作节点上并行运行。我尝试了以下方法。

from kafka import KafkaProducer
producer = KafkaProducer(bootstrap_servers = util.get_broker_metadata())
df = sqlContext.createDataFrame([("foo", 1), ("bar", 2), ("baz", 3)], ('k', 'v'))
def custom_fun(row):
    producer.send('topic',str(row.asDict()))
    producer.flush()

df.foreach(custom_fun)

这不会并给出酸洗错误。 PicklingError: Cannot pickle objects of type <type 'itertools.count'> 无法理解此错误背后的原因。谁能帮我理解这个错误或提供任何其他并行解决方案？

Answer 1

您收到的错误看起来与 Kafka 写入无关。看起来您在代码中的其他地方使用了itertools.count（AFAIK，它根本没有在 Spark 的源代码中使用，它当然有可能与KafkaProducer 一起提供）由于某种原因使用cloudpickle 模块序列化。更改 Kafka 编写代码可能根本没有影响。如果KafkaProducer 是错误的来源，您应该可以使用forachPartition 解决此问题：

from kafka import KafkaProducer


def send_to_kafka(rows):
    producer = KafkaProducer(bootstrap_servers = util.get_broker_metadata())
    for row in rows:
        producer.send('topic',str(row.asDict()))  
        producer.flush()

df.foreachPartition(send_to_kafka)

话虽如此：

或提供任何其他并行解决方案？

我建议使用 Kafka 源。包含Kafka SQL包，例如：

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

还有：

from pyspark.sql.functions import to_json, col, struct

(df 
    .select(to_json(struct([col(c).alias(c) for c in df.columns])))
    .write
    .format("kafka") 
    .option("kafka.bootstrap.servers", botstrap_servers) 
    .option("topic", topic)
    .save())