title: Use schema to convert AVRO messages with Spark to DataFrame
body: <p>Is there a way to use a schema to convert <a href="/questions/tagged/avro" class="post-tag" title="show questions tagged &#39;avro&#39;" rel="tag">avro</a> messages from <a href="/questions/tagged/kafka" class="post-tag" title="show questions tagged &#39;kafka&#39;" rel="tag">kafka</a> with <a href="/questions/tagged/spark" class="post-tag" title="show questions tagged &#39;spark&#39;" rel="tag">spark</a> to <a href="/questions/tagged/dataframe" class="post-tag" title="show questions tagged &#39;dataframe&#39;" rel="tag">dataframe</a>? The schema file for user records:</p>
  
  <pre><code>{
    "fields": [
      { "name": "firstName", "type": "string" },
      { "name": "lastName", "type": "string" }
    ],
    "name": "user",
    "type": "record"
  }
  </code></pre>
  
  <p>And code snippets from <a href="https://github.com/apache/spark/blob/master/examples/src/main/scala/org/apache/spark/examples/streaming/SqlNetworkWordCount.scala" rel="noreferrer">SqlNetworkWordCount example</a> and <a href="http://aseigneurin.github.io/2016/03/04/kafka-spark-avro-producing-and-consuming-avro-messages.html" rel="noreferrer">Kafka, Spark and Avro - Part 3, Producing and consuming Avro messages</a> to read in messages. </p>
  
  <pre class="lang-scala prettyprint-override"><code>object Injection {
    val parser = new Schema.Parser()
    val schema = parser.parse(getClass.getResourceAsStream("/user_schema.json"))
    val injection: Injection[GenericRecord, Array[Byte]] = GenericAvroCodecs.toBinary(schema)
  }
  
  ...
  
  messages.foreachRDD((rdd: RDD[(String, Array[Byte])]) =&gt; {
    val sqlContext = SQLContextSingleton.getInstance(rdd.sparkContext)
    import sqlContext.implicits._
  
    val df = rdd.map(message =&gt; Injection.injection.invert(message._2).get)
      .map(record =&gt; User(record.get("firstName").toString, records.get("lastName").toString)).toDF()
  
    df.show()
  })
  
  case class User(firstName: String, lastName: String)
  </code></pre>
  
  <p>Somehow I can't find another way than using a case class to convert AVRO messages to DataFrame. Is there a possibility to use the schema instead? I'm using <code>Spark 1.6.2</code> and <code>Kafka 0.10</code>.</p>
  
  <p>The complete code, in case you're interested.</p>
  
  <pre class="lang-scala prettyprint-override"><code>import com.twitter.bijection.Injection
  import com.twitter.bijection.avro.GenericAvroCodecs
  import kafka.serializer.{DefaultDecoder, StringDecoder}
  import org.apache.avro.Schema
  import org.apache.avro.generic.GenericRecord
  import org.apache.spark.rdd.RDD
  import org.apache.spark.sql.SQLContext
  import org.apache.spark.streaming.kafka._
  import org.apache.spark.streaming.{Seconds, StreamingContext, Time}
  import org.apache.spark.{SparkConf, SparkContext}
  
  object ReadMessagesFromKafka {
    object Injection {
      val parser = new Schema.Parser()
      val schema = parser.parse(getClass.getResourceAsStream("/user_schema.json"))
      val injection: Injection[GenericRecord, Array[Byte]] = GenericAvroCodecs.toBinary(schema)
    }
  
    def main(args: Array[String]) {
      val brokers = "127.0.0.1:9092"
      val topics = "test"
  
      // Create context with 2 second batch interval
      val sparkConf = new SparkConf().setAppName("ReadMessagesFromKafka").setMaster("local[*]")
      val ssc = new StreamingContext(sparkConf, Seconds(2))
  
      // Create direct kafka stream with brokers and topics
      val topicsSet = topics.split(",").toSet
      val kafkaParams = Map[String, String]("metadata.broker.list" -&gt; brokers)
      val messages = KafkaUtils.createDirectStream[String, Array[Byte], StringDecoder, DefaultDecoder](
    ssc, kafkaParams, topicsSet)
  
      messages.foreachRDD((rdd: RDD[(String, Array[Byte])]) =&gt; {
        val sqlContext = SQLContextSingleton.getInstance(rdd.sparkContext)
        import sqlContext.implicits._
  
        val df = rdd.map(message =&gt; Injection.injection.invert(message._2).get)
      .map(record =&gt; User(record.get("firstName").toString, records.get("lastName").toString)).toDF()
  
        df.show()
      })
  
      // Start the computation
      ssc.start()
      ssc.awaitTermination()
    }
  }
  
  /** Case class for converting RDD to DataFrame */
  case class User(firstName: String, lastName: String)
  
  /** Lazily instantiated singleton instance of SQLContext */
  object SQLContextSingleton {
    @transient  private var instance: SQLContext = _
  
    def getInstance(sparkContext: SparkContext): SQLContext = {
      if (instance == null) {
        instance = new SQLContext(sparkContext)
      }
      instance
    }
  }
  </code></pre>