Why are you calling foreachRdd / collect in the first place? Instead of using a custom decoder, you should simply do – this is code executed on the workers and allows the computation to continue. ForeachRdd and collect are output operations and force the data to be collected on the driver (assuming you don’t want that…)
val events = kafkaDStream.map { case(devId,byteArray)=> KafkaGenericEvent.parseFrom(byteArray) } From: srungarapu vamsi Date: Thursday, September 17, 2015 at 4:03 PM To: user Subject: Spark Streaming kafka directStream value decoder issue I am using KafkaUtils.createDirectStream to read the data from kafka bus. On the producer end, i am generating in the following way: 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, KafkaGenericEvent](props) // Send some messages println("Sending message") val kafkaGenericEvent = new KafkaGenericEvent("event-id",EventType.site,"6",1440500400000L) val message = new ProducerRecord[String, KafkaGenericEvent](topic,"myKey", kafkaGenericEvent) producer.send(message) } I am connecting to kafka using the console consumer script and am able to see proper data. The KafkaGenericEvent used in the above code is the class generated using ScalaBuff from a protobuff file. On the consumer end, If i read the value as a normal byte array and the convert it into KafkaGenericEvent in the following way, i get proper data: val kafkaDStream = KafkaUtils.createDirectStream[String,Array[Byte],StringDecoder,DefaultDecoder](ssc, kafkaConf, Set(topics)) kafkaDStream.foreachRDD(rdd=>rdd.collect().map{ case(devId,byteArray)=>{ println(KafkaGenericEvent.parseFrom(byteArray)) } }) But if change the value to KafkaGenericEvent and use a custom decoder like this: class KafkaGenericEventsDecoder(props: VerifiableProperties = null) extends Decoder[KafkaGenericEvent]{ override def fromBytes(bytes:Array[Byte]):KafkaGenericEvent = { KafkaGenericEvent.parseFrom(bytes) } } and in consumer: val kafkaDStream = KafkaUtils.createDirectStream[String,KafkaGenericEvent,StringDecoder,KafkaGenericEventsDecoder](ssc, kafkaConf, Set(topics)) kafkaDStream foreachRDD(rdd=>rdd.collect().map{ case(devId,genericEvent)=>{ println(genericEvent) } }) Now, i my value object KafkaGenericEvent is not created based on the sent data instead it is creating an empty Object of KafkaGenericEvent with default values. Even if i read the value as array of bytes in the createDirectStream and than apply a transformation in the following way i am getting in correct values: val kafkaDStream = KafkaUtils.createDirectStream[String,Array[Byte],StringDecoder,DefaultDecoder](ssc, kafkaConf, Set(topics)) kafkaDStream.map{ case(devId,byteArray) =>(devId,KafkaGenericEvent.parseFrom(byteArray)) } foreachRDD(rdd=>rdd.collect().map{ case(devId,genericEvent)=>{ println(genericEvent) } }) I get the default KafkaGenericEvent Object in the line println(genericEvent) Does this mean that I can transform the values only on the driver and not on the executors? I am completely confused here! I am using : scala-2.10.4 spark-1.3.1 kafka_2.10-0.8.2.1 - /Vamsi