Integrate Spark with Kafka

From EEP-5.0.0, Spark can be integrated with Kafka-1.0. You can configure a Spark application to produce Kafka messages.

About this task

Procedure

  1. Add the following dependency: 
    groupId = org.apache.spark
artifactId = spark-streaming-kafka-producer_2.11
version = <spark_version>-mapr-<mapr_eco_version>
  2. When you write the Spark program, import and use classes from: 
    org.apache.spark.streaming.kafka.producer._ 
org.apache.spark.streaming.dstream.
    The import of org.apache.spark.streaming.stream.DStream adds the following method from DStream: 
    sendToKafka(topic: String, conf: ProducerConf)
  3. In the code below, calling sendToKafka will send numMessages messages to the set of topics specified by the topics parameter: 
    val producerConf = new ProducerConf(
bootstrapServers = kafkaBrokers.split(",").toList)
                        
val items = (0 until numMessages.toInt).map(i => Item(i, i).toString)
val defaultRDD: RDD[String] = ssc.sparkContext.parallelize(items)
val dStream: DStream[String] = new ConstantInputDStream[String](ssc, defaultRDD)
                        
dStream.foreachRDD(_.sendToKafka(topics, producerConf))
dStream.count().print()

Example

Source code for a sample producer program can be found at https://github.com/mapr/spark/blob/2.2.1-mapr-1803/examples/src/main/scala/org/apache/spark/examples/streaming/KafkaProducerExample.scala