title: Java Spark Streaming JSON parsing
body: <p>I have started learning spark-streaming from Spark engine and very new to data analytics and spark. I just want to create a small IOT application in which I want to forecast future data. </p>
  
  <p>I have Tiva hardware which sends realtime sensor JSON data as follows,</p>
  
  <pre><code>[{"t":1478091719000,"sensors":[{"s":"s1","d":"+253.437"},{"s":"s2","d":"+129.750"},{"s":"s3","d":"+45.500"},{"s":"s4","d":"+255.687"},{"s":"s5","d":"+290.062"},{"s":"s6","d":"+281.500"},{"s":"s7","d":"+308.250"},{"s":"s8","d":"+313.812"}]}]
  </code></pre>
  
  <p>In this t is unix time stamp for which data is posted. 
  sensors is array of sensors with each sensor('s') data as 'd'.</p>
  
  <p>What I want to do is, consume this data and create object which spark-streaming and then pass all data though spark's Mlib (machine learning) or equivalent library to forecast future data.</p>
  
  <p>I want a general idea whether this will be possible with all technology choices </p>
  
  <ol>
  <li>I have decided to use?</li>
  <li>How can I consume the nested JSON? I tried using SQLContext but got no success.</li>
  <li>General guidelines to achieve what I am trying to do here. </li>
  </ol>
  
  <p>Here is code which I am using to consume messages from KAFKA.</p>
  
  <pre><code>SparkConf conf = new SparkConf().setAppName("DattusSpark").setMaster("local[2]");
  
      JavaSparkContext sc = new JavaSparkContext(conf);
  
      JavaStreamingContext ssc = new JavaStreamingContext(sc, new Duration(2000));
  
      // TODO: processing pipeline
      Map&lt;String, String&gt; kafkaParams = new HashMap&lt;String, String&gt;();
      kafkaParams.put("metadata.broker.list", "kafkaserver_address:9092");
      Set&lt;String&gt; topics = Collections.singleton("RAH");
  
  
      JavaPairInputDStream&lt;String, String&gt; directKafkaStream = 
              KafkaUtils.createDirectStream(ssc, String.class, String.class, StringDecoder.class,
                      StringDecoder.class, kafkaParams, topics);
  
  
      JavaDStream&lt;String&gt; json = directKafkaStream.map(new Function&lt;Tuple2&lt;String,String&gt;, String&gt;() {
          public String call(Tuple2&lt;String,String&gt; message) throws Exception {
              System.out.println(message._2());
              return message._2();
          };
      });
  
  
      System.out.println(" json is  0------ 0"+ json);
  
  
  
      json.foreachRDD(rdd -&gt; {
          rdd.foreach(
                  record -&gt; System.out.println(record));
      });
  
      ssc.start();
      ssc.awaitTermination(); 
  </code></pre>
  
  <p>PS: I want to do this in Java as to maintain linearity and good performance.</p>