CODEWITHSUNDEEP
scalaapache-sparkapache-kafkakafka-consumer-apikafka-producer-api
Vishal Sharma
Vishal Sharma

Reputation: 175

Unable to create Kafka Producer Object in Intellij

I am trying my hands on Kafka in Intellij using Spark & Scala. While creating producer Object I am unable to rectify the error. The code in Scala object is given below:

import java.util.Properties 
import org.apache.kafka.clients.producer._ 
import kafka.producer.KeyedMessage 
import org.apache.spark._
object kafkaProducer {

    def main(args: Array[String]){

      val topic = "jovis"

      val props = new Properties()
      props.put("metadata.broker.list", "localhost:9092")
      props.put("serializer.class", "kafka.serializer.StringEncoder")
      val config = new ProducerConfig(props)

//Error in Line below

val producer = new Producer[String, String](config)
      val conf = new SparkConf().setAppName("Kafka").setMaster("local")
      //val ssc = new StreamingContext(conf, Seconds(10))
      val sc = new SparkContext(conf)
      val data = sc.textFile("/home/hdadmin/empname.txt")

      var i = 0
      while(i <= data.count){
        data.collect().foreach(x => {
          println(x)
          producer.send(new KeyedMessage[String, String](topic, x))
          Thread.sleep(1000)
        })
      }

Error Log:

constructor ProducerConfig in class ProducerConfig cannot be accessed in object kafkaProducer
val config = new ProducerConfig(props)

Trait Producer is abstract;Cannot be instantiated.
val producer = new Producer[String, String](config)

I have imported the dependency jars below: http://central.maven.org/maven2/org/apache/kafka/kafka-clients/0.8.2.0/kafka-clients-0.8.2.0.jar http://central.maven.org/maven2/org/apache/kafka/kafka_2.11/0.10.2.1/kafka_2.11-0.10.2.1.jar

Apart from that I have started zookeeper server as well.

Where am I going wrong?

Upvotes: 0

Views: 1361

Answers (1)

Ivan
Ivan

Reputation: 462

May be this will help you what is the difference between kafka ProducerRecord and KeyedMessage

Please, try the new API "org.apache.kafka" %% "kafka" % "0.8.2.0"

import org.apache.kafka.clients.producer.ProducerRecord
import org.apache.kafka.clients.producer.KafkaProducer
val producer = new KafkaProducer[String, String](props)
producer.send(new ProducerRecord[String, String](topic, key, value)

Upvotes: 1

Related Questions