title: How to parse Json formatted Kafka message in spark streaming
body: <p>I have JSON messages on Kafka like this:</p>
  
  <blockquote>
    <p><code>{"id_post":"p1", "message":"blablabla"}</code></p>
  </blockquote>
  
  <p>and I want to parse the message, and print (or use for further computation) the <code>message</code> element.
  With the following code I print the json </p>
  
  <pre><code>val kafkaStream = KafkaUtils.createStream(ssc, zkQuorum, inputGroup, topicMap)
  val postStream = kafkaStream.map(_._2)
  postStream.foreachRDD((rdd, time) =&gt; {
  val count = rdd.count()
  if (count &gt; 0){
      rdd.foreach(record =&gt; {
        println(record)
      }
  }
  </code></pre>
  
  <p>but I can't manage to get the single element. 
  I tried a few JSON parser, but no luck. 
  Any idea?</p>
  
  <blockquote>
    <p>update:
    a few errors with different JSON parser
    this is the code and output with <a href="https://github.com/travisbrown/circe" rel="nofollow">circe</a> parser:</p>
  </blockquote>
  
  <pre><code>val parsed_record = parse(record)
  </code></pre>
  
  <p>and the output:</p>
  
  <pre><code>14:45:00,676 ERROR Executor:95 - Exception in task 0.0 in stage 4.0 (TID 4)
  java.lang.NoSuchMethodError: scala.Predef$.ArrowAssoc(Ljava/lang/Object;)Ljava/lang/Object;
          at io.circe.jawn.CirceSupportParser$$anon$1$$anon$4.add(CirceSupportParser.scala:36)
          at jawn.CharBasedParser$class.parseString(CharBasedParser.scala:90)
          at jawn.StringParser.parseString(StringParser.scala:15)
          at jawn.Parser.rparse(Parser.scala:397)
          at jawn.Parser.parse(Parser.scala:338)
          at jawn.SyncParser.parse(SyncParser.scala:24)
          at jawn.SupportParser$$anonfun$parseFromString$1.apply(SupportParser.scala:15)
  </code></pre>
  
  <p>and so on.. at the row in which I use <code>parse(record)</code> 
  looks like it can't access and/or parse the string <code>record</code>.</p>
  
  <p>Same if I use <a href="https://github.com/lift/lift/tree/master/framework/lift-base/lift-json" rel="nofollow">lift-json</a> 
  at the <code>parse(record)</code> the error output is more or less the same:</p>
  
  <pre><code>16:58:20,425 ERROR Executor:95 - Exception in task 0.0 in stage 4.0 (TID 4)
  java.lang.NoSuchMethodError: scala.runtime.ObjectRef.create(Ljava/lang/Object;)Lscala/runtime/ObjectRef;
  at net.liftweb.json.JsonParser$$anonfun$2.apply(JsonParser.scala:144)
  at net.liftweb.json.JsonParser$$anonfun$2.apply(JsonParser.scala:141)
  at net.liftweb.json.JsonParser$.parse(JsonParser.scala:80)
  at net.liftweb.json.JsonParser$.parse(JsonParser.scala:45)
  at net.liftweb.json.package$.parse(package.scala:40)
  at SparkConsumer$$anonfun$main$1$$anonfun$apply$1.apply(SparkConsumer.scala:98)
  at SparkConsumer$$anonfun$main$1$$anonfun$apply$1.apply(SparkConsumer.scala:95)
  at scala.collection.Iterator$class.foreach(Iterator.scala:727)
  </code></pre>
  
  <p>at scala.collection.AbstractIterator.foreach(Iterator.scala:1157)</p>