Java Spark Streaming JSON parsing
<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>