Shivansh Srivastava Shivansh Srivastava - 10 days ago 5
Scala Question

Reading Data from Kafka from begininning using Spark?

I am trying to read data from Kafka .

This is my code:

object KafkaConsumer {

import ApplicationContext._

def main(args: Array[String]) = {

val kafkaParams = Map[String, Object](
"bootstrap.servers" -> "localhost:9092",
"key.deserializer" -> classOf[StringDeserializer],
"value.deserializer" -> classOf[StringDeserializer],
"group.id" -> s"${UUID.randomUUID().toString}",
"auto.offset.reset" -> "earliest",
"enable.auto.commit" -> (false: java.lang.Boolean)
)

val topics = Array("pressure")
val stream = KafkaUtils.createDirectStream[String, String](
streamingContext,
PreferConsistent,
Subscribe[String, String](topics, kafkaParams)
)
stream.print()
stream.map(record => (record.key, record.value)).count().print()
streamingContext.start()
}
}


It displays nothing When I run this.

To Check if data is actually present in Kafka , I tried to use the command line approach and it displays data :

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic pressure --from-beginning


Output:

TimeStamp:07/13/16 15:20:45:226769,{'Pressure':'834'}
TimeStamp:07/13/16 15:20:45:266287,{'Pressure':'855'}
TimeStamp:07/13/16 15:20:45:305694,{'Pressure':'837'}
TimeStamp:07/13/16 15:20:45:344650,{'Pressure':'834'}
TimeStamp:07/13/16 15:20:45:384191,{'Pressure':'854'}
TimeStamp:07/13/16 15:20:45:423149,{'Pressure':'821'}
TimeStamp:07/13/16 15:20:45:462579,{'Pressure':'832'}
TimeStamp:07/13/16 15:20:45:501931,{'Pressure':'843'}
TimeStamp:07/13/16 15:20:45:541074,{'Pressure':'818'}
TimeStamp:07/13/16 15:20:45:580467,{'Pressure':'835'}
TimeStamp:07/13/16 15:20:45:619704,{'Pressure':'841'}
TimeStamp:07/13/16 15:20:45:659011,{'Pressure':'823'}
TimeStamp:07/13/16 15:20:45:698307,{'Pressure':'840'}
TimeStamp:07/13/16 15:20:45:737627,{'Pressure':'840'}
TimeStamp:07/13/16 15:20:45:776708,{'Pressure':'823'}
TimeStamp:07/13/16 15:20:45:816135,{'Pressure':'858'}
TimeStamp:07/13/16 15:20:45:855531,{'Pressure':'840'}
TimeStamp:07/13/16 15:20:45:894603,{'Pressure':'831'}
TimeStamp:07/13/16 15:20:45:934156,{'Pressure':'855'}
TimeStamp:07/13/16 15:20:45:973497,{'Pressure':'816'}
TimeStamp:07/13/16 15:20:46:012590,{'Pressure':'835'}


I can't understand what's the problem exactly ?
What am I missing ?

Answer

You're missing streamingContext.awaitTermination().