我正在尝试学习Spark和Spark流使用java。开发物联网应用。我有一个kafka服务器，它接受json数据，并且我能够使用sqlcontext和foreach函数解析它。
数据格式如下：，

[{"t":1481368346000,"sensors":[{"s":"s1","d":"+149.625"},{"s":"s2","d":"+23.062"},{"s":"s3","d":"+16.375"},{"s":"s4","d":"+235.937"},{"s":"s5","d":"+271.437"},{"s":"s6","d":"+265.937"},{"s":"s7","d":"+295.562"},{"s":"s8","d":"+301.687"}]}]

其中，t是每个数据流的时间戳，传感器是传感器数据的数组，s是每个传感器的名称，d包含数据。
我现在所做的是，

JavaPairInputDStream<String, String> directKafkaStream = 
            KafkaUtils.createDirectStream(ssc,
                                          String.class,
                                          String.class, 
                                          StringDecoder.class,
                                          StringDecoder.class,
                                          kafkaParams, 
                                          topics);

    SQLContext sqlContext = spark.sqlContext();
    StreamingLinearRegressionWithSGD model = new StreamingLinearRegressionWithSGD().setInitialWeights(Vectors.zeros(2));

    JavaDStream<String> json = directKafkaStream.map(new Function<Tuple2<String,String>, String>() {
        public String call(Tuple2<String,String> message) throws Exception {
            return message._2();
        };
    });
    json.print();

    json.foreachRDD(new VoidFunction<JavaRDD<String>>() {
        @Override
        public void call(JavaRDD<String> jsonRecord) throws Exception {
            System.out.println("JSON Record ---- "+jsonRecord);
             if(!jsonRecord.isEmpty()){
                    Dataset<Row> timestamp = sqlContext.read().json(jsonRecord).select("t");
                    timestamp.printSchema();
                    timestamp.show(false);

                    Dataset<Row> data = sqlContext.read().json(jsonRecord).select("sensors");
                    data.printSchema();
                    data.show(false);
                    //DF in table
                    Dataset<Row> df = data.select(org.apache.spark.sql.functions.explode(org.apache.spark.sql.functions.col("sensors")))
                            .toDF("sensors").select("sensors.s","sensors.d").where("sensors.s = 's1'");
                    Row firstRow = df.head();

                    String valueOfFirstSensor = firstRow.getString(1);
                    System.out.println("---------valueOfFirstSensor --------"+ valueOfFirstSensor);
                    double[] values = new double[1];
                    values[0] = firstRow.getDouble(0);
                    new LabeledPoint(timestamp.head().getDouble(0), Vectors.dense(values));

                    df.show(false);
                }
        }
    });

    ssc.start();
    ssc.awaitTermination();

我想做的是，将javadstream中的json转换为StreamingLinearGressionWithGD模型接受的数据结构。
当我尝试使用sparks的map函数将json流Map到javadstream时，如下所示，

JavaDStream<LabeledPoint> forML =  json.map(new Function<String, LabeledPoint>() {

            @Override
            public LabeledPoint call(String jsonRecord) throws Exception {
                // TODO Auto-generated method stub
                System.out.println("\n\n\n here is JSON in"+ jsonRecord);

                LabeledPoint returnObj = null;
                 if(!jsonRecord.isEmpty()){
                        Dataset<Row> timestamp = sqlContext.read().json(jsonRecord).select("t");
                        timestamp.printSchema();
                        timestamp.show(false);

                        Dataset<Row> data = sqlContext.read().json(jsonRecord).select("sensors");
                        data.printSchema();
                        data.show(false);
                        //DF in table
                        Dataset<Row> df = data.select(org.apache.spark.sql.functions.explode(org.apache.spark.sql.functions.col("sensors")))
                                .toDF("sensors").select("sensors.s","sensors.d").where("sensors.s = 's1'");
                        Row firstRow = df.head();

                        String valueOfFirstSensor = firstRow.getString(1);
                        System.out.println("---------valueOfFirstSensor --------"+ valueOfFirstSensor);
                        double[] values = new double[1];
                        values[0] = firstRow.getDouble(0);
                        returnObj = new LabeledPoint(timestamp.head().getDouble(0), Vectors.dense(values));
                        df.show(false);
                    }
                 return returnObj;

            }
        }).cache();

model.trainOn(forML);

并调用model.train，但失败时出现nullpointerexception

Dataset<Row> timestamp = sqlContext.read().json(jsonRecord).select("t");

现在我的问题是，
我做得对吗？
我将如何预测值，为什么以及如何需要创建一个不同的流来将其传递给模型的predicton函数？
我将接收多个传感器，但每个传感器只有一个值，而且可能有成千上万个这样的流，我如何才能为这几千个传感器中的每一个创建不同的模型，并有效地预测如此大量的数据？
有没有其他好的机器学习算法或方法可以用于这种类型的传感器数据？