Yes, unfortunately Spark (1.x, 2.x) doesn't make it straight-forward how to write to Kafka in an efficient manner.

I'd suggest the following approach:

• Use (and re-use) one KafkaProducer instance per executor process/JVM.

Here's the high-level setup for this approach:

1. First, you must "wrap" Kafka's KafkaProducer because, as you mentioned, it is not serializable. Wrapping it allows you to "ship" it to the executors. The key idea here is to use a lazy val so that you delay instantiating the producer until its first use, which is effectively a workaround so that you don't need to worry about KafkaProducer not being serializable.


2. You "ship" the wrapped producer to each executor by using a broadcast variable.


3. Within your actual processing logic, you access the wrapped producer through the broadcast variable, and use it to write processing results back to Kafka.

The code snippets below work with Spark Streaming as of Spark 2.0.

Step 1: Wrapping KafkaProducer

import java.util.concurrent.Future



import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord, RecordMetadata}



class MySparkKafkaProducer[K, V](createProducer: () => KafkaProducer[K, V]) extends Serializable {



  /* This is the key idea that allows us to work around running into

     NotSerializableExceptions. */

  lazy val producer = createProducer()



  def send(topic: String, key: K, value: V): Future[RecordMetadata] =

    producer.send(new ProducerRecord[K, V](topic, key, value))



  def send(topic: String, value: V): Future[RecordMetadata] =

    producer.send(new ProducerRecord[K, V](topic, value))



}



object MySparkKafkaProducer {



  import scala.collection.JavaConversions._



  def apply[K, V](config: Map[String, Object]): MySparkKafkaProducer[K, V] = {

    val createProducerFunc = () => {

      val producer = new KafkaProducer[K, V](config)



      sys.addShutdownHook {

        // Ensure that, on executor JVM shutdown, the Kafka producer sends

        // any buffered messages to Kafka before shutting down.

        producer.close()

      }



      producer

    }

    new MySparkKafkaProducer(createProducerFunc)

  }



  def apply[K, V](config: java.util.Properties): MySparkKafkaProducer[K, V] = apply(config.toMap)



}

Step 2: Use a broadcast variable to give each executor its own wrapped KafkaProducer instance

import org.apache.kafka.clients.producer.ProducerConfig



val ssc: StreamingContext = {

  val sparkConf = new SparkConf().setAppName("spark-streaming-kafka-example").setMaster("local[2]")

  new StreamingContext(sparkConf, Seconds(1))

}



ssc.checkpoint("checkpoint-directory")



val kafkaProducer: Broadcast[MySparkKafkaProducer[Array[Byte], String]] = {

  val kafkaProducerConfig = {

    val p = new Properties()

    p.setProperty("bootstrap.servers", "broker1:9092")

    p.setProperty("key.serializer", classOf[ByteArraySerializer].getName)

    p.setProperty("value.serializer", classOf[StringSerializer].getName)

    p

  }

  ssc.sparkContext.broadcast(MySparkKafkaProducer[Array[Byte], String](kafkaProducerConfig))

}

Step 3: Write from Spark Streaming to Kafka, re-using the same wrapped KafkaProducer instance (for each executor)

import java.util.concurrent.Future

import org.apache.kafka.clients.producer.RecordMetadata



val stream: DStream[String] = ???

stream.foreachRDD { rdd =>

  rdd.foreachPartition { partitionOfRecords =>

    val metadata: Stream[Future[RecordMetadata]] = partitionOfRecords.map { record =>

      kafkaProducer.value.send("my-output-topic", record)

    }.toStream

    metadata.foreach { metadata => metadata.get() }

  }

}

Hope this helps.

There is a Streaming Kafka Writer maintained by Cloudera (actually spun off from a Spark JIRA [1]). It basically creates a producer per partition, which amortizes the time spent to create 'heavy' objects over a (hopefully large) collection of elements.

The Writer can be found here: https://github.com/cloudera/spark-kafka-writer

I was having the same issue and found this post.

The author solves the problem by creating 1 producer per executor. Instead of sending the producer itself, he sends only a “recipe” how to create a producer in an executor by broadcasting it.

    val kafkaSink = sparkContext.broadcast(KafkaSink(conf))

He uses a wrapper that lazily creates the producer:

    class KafkaSink(createProducer: () => KafkaProducer[String, String]) extends Serializable {



      lazy val producer = createProducer()



      def send(topic: String, value: String): Unit = producer.send(new     ProducerRecord(topic, value))

    }





    object KafkaSink {

      def apply(config: Map[String, Object]): KafkaSink = {

        val f = () => {

          val producer = new KafkaProducer[String, String](config)



          sys.addShutdownHook {

            producer.close()

          }



          producer

        }

        new KafkaSink(f)

      }

    }

The wrapper is serializable because the Kafka producer is initialized just before first use on an executor. The driver keeps the reference to the wrapper and the wrapper sends the messages using each executor's producer:

    dstream.foreachRDD { rdd =>

      rdd.foreach { message =>

        kafkaSink.value.send("topicName", message)

      }

    }

Why is it infeasible? Fundamentally each partition of each RDD is going to run independently (and may well run on a different cluster node), so you have to redo the connection (and any synchronization) at the start of each partition's task. If the overhead of that is too high then you should increase the batch size in your StreamingContext until it becomes acceptable (obv. there's a latency cost to doing this).

(If you're not handling thousands of messages in each partition, are you sure you need spark-streaming at all? Would you do better with a standalone application?)

With Spark >= 2.2

Both read and write operations are possible on Kafka using Structured Streaming API

Build stream from Kafka topic

// Subscribe to a topic and read messages from the earliest to latest offsets

val ds= spark

  .readStream // use `read` for batch, like DataFrame

  .format("kafka")

  .option("kafka.bootstrap.servers", "brokerhost1:port1,brokerhost2:port2")

  .option("subscribe", "source-topic1")

  .option("startingOffsets", "earliest")

  .option("endingOffsets", "latest")

  .load()

Read the key and value and apply the schema for both, for simplicity we are making converting both of them to String type.

val dsStruc = ds.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")

  .as[(String, String)]

Since dsStruc have the schema, it's accepts all SQL kind operations like filter, agg, select ..etc on it.

Write stream to Kafka topic

dsStruc

  .writeStream // use `write` for batch, like DataFrame

  .format("kafka")

  .option("kafka.bootstrap.servers", "brokerhost1:port1,brokerhost2:port2")

  .option("topic", "target-topic1")

  .start()

More configuration for Kafka integration to read or write

Key artifacts to add in the application

 "org.apache.spark" % "spark-core_2.11" % 2.2.0,

 "org.apache.spark" % "spark-streaming_2.11" % 2.2.0,

 "org.apache.spark" % "spark-sql-kafka-0-10_2.11" % 2.2.0,

This might be what you want to do. You basically create one producer for each partition of records.

input.foreachRDD(rdd =>

      rdd.foreachPartition(

          partitionOfRecords =>

            {

                val props = new HashMap[String, Object]()

                props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers)

                props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,

                  "org.apache.kafka.common.serialization.StringSerializer")

                props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,

                  "org.apache.kafka.common.serialization.StringSerializer")

                val producer = new KafkaProducer[String,String](props)



                partitionOfRecords.foreach

                {

                    case x:String=>{

                        println(x)



                        val message=new ProducerRecord[String, String]("output",null,x)

                        producer.send(message)

                    }

                }

          })

) 

Hope that helps