KafkaInput.scala

package com.powerpanel.kso;

import kafka.serializer.StringDecoder
import kafka.serializer.DefaultDecoder
import com.powerpanel.kso.model._
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.kafka010.ConsumerStrategies
import org.apache.spark.streaming.kafka010.LocationStrategies
import org.apache.avro.specific.SpecificDatumReader
import org.apache.avro.file.{DataFileReader, SeekableByteArrayInput}
// import scala.collection.JavaConverters._
import org.apache.spark.streaming.StreamingContext
import org.apache.avro.io.DecoderFactory
class KafkaInput extends Serializable {
    def readFromKafka (ssc: StreamingContext) = {
        val props = AppConfig.loadProperties();
        val topicsSet = props.getProperty("kafka.topic").split(",").toSet;
        
	val kafkaParams = collection.mutable.Map[String, String]("bootstrap.servers" -> props.getProperty("kafka.brokers"))
        if (props.getProperty("kafka.consume_from_beginning").toBoolean)
        {
            kafkaParams.put("auto.offset.reset", "earliest");
        }
        kafkaParams.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        kafkaParams.put("value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");
	      kafkaParams.put("group.id", props.getProperty("kafka.group_id"));
        val messages = KafkaUtils.createDirectStream[String, Array[Byte]](
          ssc,
          LocationStrategies.PreferConsistent,
          ConsumerStrategies.Subscribe[String, Array[Byte]](topicsSet, kafkaParams)
        );
        //.repartition(Integer.parseInt(props.getProperty("app.processing_parallelism")));

        // Decode avro container format
        // val avroSchemaString = StaticHelpers.loadResourceFile("dataplatform-raw.avsc");
        val rawMessages = messages.map(x => {
            // val eventDecoder = new DataPlatformEventDecoder(avroSchemaString);
            // val list = dataFileReader.iterator().asScala.toList
            // DataFileReader<User> dataFileReader = new DataFileReader<User>(file, userDatumReader);
            // val dataPackageDecoder = DataPackage.getDecoder();
            val payload = x.value();
            /*
            val datumReader = new SpecificDatumReader[DataPackage](DataPackage.getClassSchema)
            val inputStream = new SeekableByteArrayInput(payload)
            val dataFileReader = new DataFileReader[DataPackage](inputStream, datumReader)
            val dataPackage = dataFileReader.next()*/
            val datumReader = new SpecificDatumReader[DataPackage](DataPackage.getClassSchema)
            val inputStream = new SeekableByteArrayInput(payload)
            val decoder = DecoderFactory.get.binaryDecoder(inputStream, null)
          
            // while (!decoder.isEnd) {
              val dataPackage = datumReader.read(null, decoder)
            //}
            // val dataPackage = dataPackageDecoder.decode(payload);
            dataPackage;
        });
        rawMessages;
    };
}