title: spark Streaming avro data from kafka topic with scala
body: <p>I've been trying to stream avro data from kafka topics using spark streaming written in scala.(<a href="https://community.hortonworks.com/articles/33275/receiving-avro-messages-through-kafka-in-a-spark-s.html" rel="nofollow noreferrer">i followed this link</a>).
  The main motto is to parse avro data from kafka topic and create a data frame out of the dstreams and persist in to hdfs.
  The code runs fine without any exception but there is no output as well.</p>
  
  <pre><code>case class mobileData(action: String, tenantid: Int, lat: Float, lon: Float, memberid: Int, event_name: String, productUpccd: Int, device_type: String, device_os_ver: Float, item_name: String)
  object AvroConsumer {
    val eventSchema = SchemaBuilder.record("eventRecord").fields
      .name("action").`type`().stringType().noDefault()
      .name("tenantid").`type`().intType().noDefault()
      .name("lat").`type`().doubleType().noDefault()
      .name("lon").`type`().doubleType().noDefault()
      .name("memberid").`type`().intType().noDefault()
      .name("event_name").`type`().stringType().noDefault()
      .name("productUpccd").`type`().intType().noDefault()
      .name("device_type").`type`().stringType().noDefault()
      .name("device_os_ver").`type`().stringType().noDefault()
      .name("item_name").`type`().stringType().noDefault().endRecord
    def main(args: Array[String]): Unit = {
      Logger.getLogger("kafka.consumer.AvroConsumer").setLevel(Level.WARN)
      Logger.getLogger("org.apache.spark.storage.BlockManager").setLevel(Level.ERROR)
      val logger: Logger = Logger.getLogger("kafka.consumer.AvroConsumer")
      val sparkConf = new SparkConf().setAppName("Avro Consumer").
        setMaster("dev-mapr-node2.fishbowl.com").set("spark.driver.allowMultipleContexts", "true")
      sparkConf.set("spark.cores.max", "2")
      sparkConf.set("spark.serializer", classOf[KryoSerializer].getName)
      sparkConf.set("spark.sql.tungsten.enabled", "true")
      sparkConf.set("spark.eventLog.enabled", "true")
      sparkConf.set("spark.app.id", "KafkaConsumer")
      sparkConf.set("spark.io.compression.codec", "snappy")
      sparkConf.set("spark.rdd.compress", "true")
      sparkConf.set("spark.streaming.backpressure.enabled", "true")
      sparkConf.set("spark.sql.avro.compression.codec", "snappy")
      sparkConf.set("spark.sql.avro.mergeSchema", "true")
      sparkConf.set("spark.sql.avro.binaryAsString", "true")
      val sc = new SparkContext(sparkConf)
      sc.hadoopConfiguration.set("avro.enable.summary-metadata", "false")
      val ssc = new StreamingContext(sc, Seconds(2))
      try {
        val kafkaConf = Map[String, String]("metadata.broker.list" -&gt; "#################:9092",
          "zookeeper.connect" -&gt; "###################:5181",
          "group.id" -&gt; "KafkaConsumer",
          "zookeeper.connection.timeout.ms" -&gt; "1000000")
        val topicMaps = Map("fishbowl" -&gt; 1)
        val messages: ReceiverInputDStream[(String, Array[Byte])] = KafkaUtils.createStream[String, Array[Byte], DefaultDecoder, DefaultDecoder](ssc, kafkaConf, topicMaps, StorageLevel.MEMORY_ONLY_SER)
        try {
          messages.foreachRDD((rdd, time) =&gt; {
            if (rdd != null) {
              try {
                import com.databricks.spark.avro._
                import org.apache.spark.sql.SQLContext
                val sqlContext = new SQLContext(sc)
                import sqlContext.implicits._
  
                val rdd2 = rdd.map { case (k, v) =&gt; parseAVROToString(v) }
  
                try {
                  val result = rdd2.mapPartitions(records =&gt; {
                    val mapper = new ObjectMapper()
                    mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
                    mapper.registerModule(DefaultScalaModule)
                    records.flatMap(record =&gt; {
                      try {
                        Some(mapper.readValue(record, classOf[mobileData]))
                      } catch {
                        case e: Exception =&gt; None;
                      }
  
                    })
                  }, true)
                  println(result)
  
                  val df1 = result.toDF()
                  df1.printSchema()
                  df1.show()
                  logger.error("Registered Events: " + df1.count())
                  df1.registerTempTable("mdl_events")
  
                  df1.write.format("com.databricks.spark.avro").mode(org.apache.spark.sql.SaveMode.Append).save("/tmp/output")
                  //df1.write.format("org.apache.sql").mode(SaveMode.Append).save("/tmp/output")
                  //df1.write.format("avro").mode(org.apache.spark.sql.SaveMode.Append).avro("/avroResults")
                //  sqlContext.read.format("com.databricks.spark.avro").load("/avroResults")
                } catch {
                  case e: Exception =&gt; None;
                }
              } catch {
                case e: Exception =&gt; None;
              }
            }
          })
        } catch {
          case e: Exception =&gt;
            println("Writing files after job. Exception:" + e.getMessage);
            e.printStackTrace();
        }
      } catch {
        case e: Exception =&gt;
          println("Kafka Stream. Writing files after job. Exception:" + e.getMessage);
          e.printStackTrace();
      }
      ssc.start()
      ssc.awaitTermination()
    }
  
    def parseAVROToString(rawEvent: Array[Byte]): String = {
      try {
        if (rawEvent.isEmpty) {
          println("Rejected Event")
          "Empty"
        } else {
          deserializeEvent(rawEvent).get("eventRecord").toString
        }
      } catch {
        case e: Exception =&gt;
          println("Exception:" + e.getMessage);
          "Empty"
      }
    }
  
    def deserializeEvent(rawEvent: Array[Byte]): GenericRecord = {
      try {
        val reader = new GenericDatumReader[GenericRecord](eventSchema)
        val decoder = DecoderFactory.get.binaryDecoder(rawEvent, null)
        reader.read(null, decoder)
      } catch {
        case e: Exception =&gt;
          None;
          null;
      }
    }
  }
  </code></pre>
  
  <p>Here is the pom.xml </p>
  
  <p><div class="snippet" data-lang="js" data-hide="false" data-console="true" data-babel="false">
  <div class="snippet-code">
  <pre class="snippet-code-html lang-html prettyprint-override"><code>&lt;project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"&gt;
    &lt;modelVersion&gt;4.0.0&lt;/modelVersion&gt;
    &lt;groupId&gt;kafka&lt;/groupId&gt;
    &lt;artifactId&gt;consumer&lt;/artifactId&gt;
    &lt;version&gt;1.0.0.0-SNAPSHOT&lt;/version&gt;
    &lt;pluginRepositories&gt;
      &lt;pluginRepository&gt;
        &lt;id&gt;scala-tools.org&lt;/id&gt;
        &lt;name&gt;Scala-tools Maven2 Repository&lt;/name&gt;
        &lt;url&gt;http://scala-tools.org/repo-releases&lt;/url&gt;
      &lt;/pluginRepository&gt;
    &lt;/pluginRepositories&gt;
  
    &lt;dependencies&gt;
      &lt;dependency&gt;
        &lt;groupId&gt;org.scala-lang&lt;/groupId&gt;
        &lt;artifactId&gt;scala-library&lt;/artifactId&gt;
        &lt;version&gt;2.10.6&lt;/version&gt;
      &lt;/dependency&gt;
      &lt;dependency&gt;
        &lt;groupId&gt;org.apache.spark&lt;/groupId&gt;
        &lt;artifactId&gt;spark-core_2.10&lt;/artifactId&gt;
        &lt;scope&gt;provided&lt;/scope&gt;
        &lt;version&gt;1.6.1&lt;/version&gt;
      &lt;/dependency&gt;
      &lt;!-- https://mvnrepository.com/artifact/org.apache.spark/spark-streaming-kafka_2.10 --&gt;
      &lt;dependency&gt;
        &lt;groupId&gt;org.apache.spark&lt;/groupId&gt;
        &lt;artifactId&gt;spark-streaming-kafka_2.10&lt;/artifactId&gt;
        &lt;version&gt;1.6.1&lt;/version&gt;
      &lt;/dependency&gt;
      &lt;dependency&gt;
        &lt;groupId&gt;org.apache.spark&lt;/groupId&gt;
        &lt;artifactId&gt;spark-streaming-kafka-0-10_2.10&lt;/artifactId&gt;
        &lt;version&gt;2.0.0&lt;/version&gt;
      &lt;/dependency&gt;
      &lt;dependency&gt;
        &lt;groupId&gt;org.apache.spark&lt;/groupId&gt;
        &lt;artifactId&gt;spark-sql_2.10&lt;/artifactId&gt;
        &lt;version&gt;1.6.1&lt;/version&gt;
      &lt;/dependency&gt;
      &lt;!-- https://mvnrepository.com/artifact/org.apache.spark/spark-examples_2.10 --&gt;
      &lt;dependency&gt;
        &lt;groupId&gt;org.apache.spark&lt;/groupId&gt;
        &lt;artifactId&gt;spark-examples_2.10&lt;/artifactId&gt;
        &lt;version&gt;1.1.1&lt;/version&gt;
      &lt;/dependency&gt;
      &lt;dependency&gt;
        &lt;groupId&gt;org.apache.spark&lt;/groupId&gt;
        &lt;artifactId&gt;spark-streaming-kafka_2.11&lt;/artifactId&gt;
        &lt;version&gt;1.6.1&lt;/version&gt;
      &lt;/dependency&gt;
      &lt;dependency&gt;
        &lt;groupId&gt;org.apache.kafka&lt;/groupId&gt;
        &lt;artifactId&gt;kafka_2.11&lt;/artifactId&gt;
        &lt;version&gt;0.9.0.0&lt;/version&gt;
      &lt;/dependency&gt;
      &lt;dependency&gt;
        &lt;groupId&gt;org.apache.avro&lt;/groupId&gt;
        &lt;artifactId&gt;avro&lt;/artifactId&gt;
        &lt;version&gt;1.8.1&lt;/version&gt;
      &lt;/dependency&gt;
      &lt;!-- https://mvnrepository.com/artifact/com.databricks/spark-avro_2.10 --&gt;
      &lt;dependency&gt;
        &lt;groupId&gt;com.databricks&lt;/groupId&gt;
        &lt;artifactId&gt;spark-avro_2.10&lt;/artifactId&gt;
        &lt;version&gt;2.0.1&lt;/version&gt;
      &lt;/dependency&gt;
  
      &lt;dependency&gt;
        &lt;groupId&gt;com.fasterxml.jackson.module&lt;/groupId&gt;
        &lt;artifactId&gt;jackson-module-scala_2.10&lt;/artifactId&gt;
        &lt;version&gt;2.8.4&lt;/version&gt;
      &lt;/dependency&gt;
      &lt;!-- https://mvnrepository.com/artifact/com.fasterxml.jackson.core/jackson-databind --&gt;
      &lt;dependency&gt;
        &lt;groupId&gt;com.fasterxml.jackson.core&lt;/groupId&gt;
        &lt;artifactId&gt;jackson-databind&lt;/artifactId&gt;
        &lt;version&gt;2.8.5&lt;/version&gt;
      &lt;/dependency&gt;
      &lt;!-- https://mvnrepository.com/artifact/com.fasterxml/jackson-xml-databind --&gt;
      &lt;dependency&gt;
        &lt;groupId&gt;com.fasterxml&lt;/groupId&gt;
        &lt;artifactId&gt;jackson-xml-databind&lt;/artifactId&gt;
        &lt;version&gt;0.6.2&lt;/version&gt;
      &lt;/dependency&gt;
    &lt;/dependencies&gt;
  
    &lt;build&gt;
      &lt;pluginManagement&gt;
        &lt;plugins&gt;
          &lt;plugin&gt;
            &lt;groupId&gt;org.codehaus.mojo&lt;/groupId&gt;
            &lt;artifactId&gt;build-helper-maven-plugin&lt;/artifactId&gt;
            &lt;version&gt;1.7&lt;/version&gt;
            &lt;executions&gt;
              &lt;execution&gt;
                &lt;id&gt;add-source&lt;/id&gt;
                &lt;phase&gt;generate-sources&lt;/phase&gt;
                &lt;goals&gt;
                  &lt;goal&gt;add-source&lt;/goal&gt;
                &lt;/goals&gt;
                &lt;configuration&gt;
                  &lt;sources&gt;
                    &lt;source&gt;src/main/java&lt;/source&gt;
                  &lt;/sources&gt;
                &lt;/configuration&gt;
              &lt;/execution&gt;
            &lt;/executions&gt;
          &lt;/plugin&gt;
        &lt;/plugins&gt;
      &lt;/pluginManagement&gt;
      &lt;plugins&gt;
  
        &lt;!-- mixed scala/java compile --&gt;
        &lt;plugin&gt;
          &lt;groupId&gt;org.scala-tools&lt;/groupId&gt;
          &lt;artifactId&gt;maven-scala-plugin&lt;/artifactId&gt;
          &lt;version&gt;2.15.2&lt;/version&gt;
          &lt;executions&gt;
            &lt;execution&gt;
              &lt;id&gt;compile&lt;/id&gt;
              &lt;goals&gt;
                &lt;goal&gt;compile&lt;/goal&gt;
              &lt;/goals&gt;
              &lt;phase&gt;compile&lt;/phase&gt;
            &lt;/execution&gt;
            &lt;execution&gt;
              &lt;id&gt;test-compile&lt;/id&gt;
              &lt;goals&gt;
                &lt;goal&gt;testCompile&lt;/goal&gt;
              &lt;/goals&gt;
              &lt;phase&gt;test-compile&lt;/phase&gt;
            &lt;/execution&gt;
            &lt;execution&gt;
              &lt;phase&gt;process-resources&lt;/phase&gt;
              &lt;goals&gt;
                &lt;goal&gt;compile&lt;/goal&gt;
              &lt;/goals&gt;
            &lt;/execution&gt;
          &lt;/executions&gt;
        &lt;/plugin&gt;
        &lt;plugin&gt;
          &lt;groupId&gt;org.apache.maven.plugins&lt;/groupId&gt;
          &lt;artifactId&gt;maven-compiler-plugin&lt;/artifactId&gt;
          &lt;version&gt;3.6.0&lt;/version&gt;
          &lt;configuration&gt;
            &lt;source&gt;1.7&lt;/source&gt;
            &lt;target&gt;1.7&lt;/target&gt;
          &lt;/configuration&gt;
        &lt;/plugin&gt;
        &lt;!-- for fatjar --&gt;
        &lt;plugin&gt;
          &lt;groupId&gt;org.apache.maven.plugins&lt;/groupId&gt;
          &lt;artifactId&gt;maven-assembly-plugin&lt;/artifactId&gt;
          &lt;version&gt;2.4&lt;/version&gt;
          &lt;configuration&gt;
            &lt;descriptorRefs&gt;
              &lt;descriptorRef&gt;jar-with-dependencies&lt;/descriptorRef&gt;
            &lt;/descriptorRefs&gt;
          &lt;/configuration&gt;
          &lt;executions&gt;
            &lt;execution&gt;
              &lt;id&gt;assemble-all&lt;/id&gt;
              &lt;phase&gt;package&lt;/phase&gt;
              &lt;goals&gt;
                &lt;goal&gt;single&lt;/goal&gt;
              &lt;/goals&gt;
            &lt;/execution&gt;
          &lt;/executions&gt;
        &lt;/plugin&gt;
        &lt;plugin&gt;
          &lt;groupId&gt;org.apache.maven.plugins&lt;/groupId&gt;
          &lt;artifactId&gt;maven-jar-plugin&lt;/artifactId&gt;
          &lt;version&gt;3.0.2&lt;/version&gt;
          &lt;configuration&gt;
            &lt;archive&gt;
              &lt;manifest&gt;
                &lt;addClasspath&gt;true&lt;/addClasspath&gt;
                &lt;mainClass&gt;kafka.consumer.AvroConsumer1&lt;/mainClass&gt;
              &lt;/manifest&gt;
            &lt;/archive&gt;
          &lt;/configuration&gt;
        &lt;/plugin&gt;
      &lt;/plugins&gt;
  
  
    &lt;/build&gt;
  
  &lt;/project&gt;</code></pre>
  </div>
  </div>
  </p>
  
  <p><strong>And here is my console log,</strong></p>
  
  <blockquote>
    <p>root  |-- action: string (nullable = true)  |-- tenantid: integer
    (nullable = false)  |-- lat: float (nullable = false)  |-- lon: float
    (nullable = false)  |-- memberid: integer (nullable = false)  |--
    event_name: string (nullable = true)  |-- productUpccd: integer
    (nullable = false)  |-- device_type: string (nullable = true)  |--
    device_os_ver: float (nullable = false)  |-- item_name: string
    (nullable = true)</p>
    
    <p>+------+--------+---+---+--------+----------+------------+-----------+-------------+---------+ |action|tenantid|lat|lon|memberid|event_name|productUpccd|device_type|device_os_ver|item_name|
    +------+--------+---+---+--------+----------+------------+-----------+-------------+---------+
    +------+--------+---+---+--------+----------+------------+-----------+-------------+---------+</p>
    
    <p>+------+ |action|
    +------+
    +------+</p>
    
    <p>16/12/13 18:22:08 INFO scheduler.JobScheduler: Finished job streaming
    job 1481633528000 ms.0 from job set of time 1481633528000 ms 16/12/13
    18:22:08 INFO scheduler.JobScheduler: Total delay: 0.097 s for time
    1481633528000 ms (execution: 0.094 s) 16/12/13 18:22:08 INFO
    rdd.BlockRDD: Removing RDD 67 from persistence list 16/12/13 18:22:08
    INFO kafka.KafkaInputDStream: Removing blocks of RDD BlockRDD[67] at
    createStream at AvroConsumer.scala:63 of time 1481633528000 ms
    16/12/13 18:22:08 INFO scheduler.ReceivedBlockTracker: Deleting
    batches ArrayBuffer(1481633524000 ms) 16/12/13 18:22:08 INFO
    scheduler.InputInfoTracker: remove old batch metadata: 1481633524000
    ms ^C16/12/13 18:22:09 INFO streaming.StreamingContext: Invoking
    stop(stopGracefully=false) from shutdown hook 16/12/13 18:22:09 INFO
    scheduler.ReceiverTracker: Sent stop signal to all 1 receivers
    16/12/13 18:22:10 INFO scheduler.JobScheduler: Added jobs for time
    1481633530000 ms 16/12/13 18:22:10 INFO scheduler.JobScheduler:
    Starting job streaming job 1481633530000 ms.0 from job set of time
    1481633530000 ms</p>
  </blockquote>
  
  <p><strong>and here is my avro producer program in java</strong></p>
  
  <pre><code>public class KafkaAvroProducer {
  
      /* case class
       TopicData("action":"AppEvent","tenantid":1173,"lat":0.0,"lon":0.0,"memberid":55,
       "event_name":"CATEGORY_CLICK",
       "productUpccd":0,"device_type":"iPhone","device_os_ver":"10.1","item_name":"CHICKEN",*/
  
      public static final String EVENT_SCHEMA = "{" + "\"type\":\"record\","
              + "\"name\":\"eventrecord\"," + "\"fields\":["
              + "  { \"name\":\"action\", \"type\":\"string\" },"
              + "  { \"name\":\"tenantid\", \"type\":\"int\" },"
              + "  { \"name\":\"lat\", \"type\":\"double\" },"
              + "  { \"name\":\"lon\", \"type\":\"double\" },"
              + "  { \"name\":\"memberid\", \"type\":\"int\" },"
              + "  { \"name\":\"event_name\", \"type\":\"string\" },"
              + "  { \"name\":\"productUpccd\", \"type\":\"int\" },"
              + "  { \"name\":\"device_type\", \"type\":\"string\" },"
              + "  { \"name\":\"device_os_ver\", \"type\":\"string\" },"
              + "{ \"name\":\"item_name\", \"type\":\"string\" }" + "]}";
  
      public static void main(String[] args) throws InterruptedException {
          Properties props = new Properties();
          props.put("bootstrap.servers", "##########:9092");
          props.put("key.serializer",
                  "org.apache.kafka.common.serialization.StringSerializer");
          props.put("value.serializer",
                  "org.apache.kafka.common.serialization.ByteArraySerializer");
          props.put("producer.type", "async");
          Schema.Parser parser = new Schema.Parser();
          Schema schema = parser.parse(EVENT_SCHEMA);
          Injection&lt;GenericRecord,byte[]&gt; avroRecords = GenericAvroCodecs.toBinary(schema);
          KafkaProducer&lt;String, byte[]&gt; producer = new KafkaProducer&lt;&gt;(props);
          for(int i = 0; i&lt;300;i++){
              GenericData.Record avroRecord = new GenericData.Record(schema);
              setEventValues(i, avroRecord);
              byte[] bytes = avroRecords.apply(avroRecord);
              ProducerRecord&lt;String, byte[]&gt; producerRecord = new ProducerRecord&lt;String, byte[]&gt;("fishbowl",bytes);
              System.out.println(producerRecord);
              producer.send(producerRecord);
  
          }
          producer.close();
      }
  
      private static void setEventValues(int i, Record avroRecord) {
  
          avroRecord.put("action", "AppEvent");
          avroRecord.put("tenantid", i);
          avroRecord.put("lat", 0.0);
          avroRecord.put("lon", 0.0);
          avroRecord.put("memberid", 55);
          avroRecord.put("event_name", "CATEGORY_CLICK");
          avroRecord.put("productUpccd", 0);
          avroRecord.put("device_type", "iPhone");
          avroRecord.put("device_os_ver", "10.1");
          avroRecord.put("item_name", "CHICKEN");
      }
  
  }
  </code></pre>