Spark：将 KafkaProducer 广播到 Spark 流的最佳方式

Question

为了广播 KafkaProducer 来激发执行者，我创建了一个如下所示的包装器：

public class KafkaSink implements Serializable {
    private static KafkaProducer<String, String> producer = null;

    public KafkaProducer<String, String> getInstance(final Properties properties) {
        if(producer == null) {
            producer = new KafkaProducer<>(properties);
        }
        return producer;
    }

    public void close() {
        producer.close();
    }
}

并像下面这样使用它

 JavaSparkContext jsc = new JavaSparkContext(sc);
 Broadcast<KafkaSink> kafkaSinkBroadcast = jsc.broadcast(new KafkaSink()));
 dataset.toJavaRDD().foreach(row -> kafkaSinkBroadcast.getValue().getInstance(kafkaProducerProps()).send(new ProducerRecord<String, String>(topic, row.mkString(", "))))

我只是想知道这样做是否正确，或者最好的方法是什么

Answer 1

我真的可以推荐这个blog post。简而言之，您应该通过传递“配方”来为每个分区创建一个可序列化的接收器来创建 Kafka 生产者。