如何在Java中解析spark流中的复杂JSON数据

1 年，3 月 Questions & Answers 4646

我们正在开发物联网应用程序

我们从每个我们想要运行分析的设备中获得以下数据流

[{"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"}]}]

在初级阶段，我可以使用spark java代码获取模式，如下所示：

    root
     |-- sensors: array (nullable = true)
     |    |-- element: struct (containsNull = true)
     |    |    |-- d: string (nullable = true)
     |    |    |-- s: string (nullable = true)
     |-- t: long (nullable = true)

我写的代码是

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

    SQLContext sqlContext = spark.sqlContext();
    json.foreachRDD(new VoidFunction<JavaRDD<String>>() {
        @Override
        public void call(JavaRDD<String> jsonRecord) throws Exception {

            Dataset<Row> row = sqlContext.read().json(jsonRecord).toDF();
            row.createOrReplaceTempView("MyTable");
            row.printSchema();
            row.show();

            Dataset<Row> sensors = row.select("sensors");
            sensors.createOrReplaceTempView("sensors");
            sensors.printSchema();
            sensors.show();

        }
    });

这给了我一个错误，即“org.apache.spark.sql.AnalysisException:无法解析给定输入列的“sensors”：

我是spark和analytics的初学者，在java中找不到解析嵌套json的好例子

我想要实现的是，可能需要专家们的建议

我将提取每个传感器值，然后使用sparkML spark库运行回归分析。这将帮助我找出每个传感器流中正在发生的趋势，以及我希望使用这些数据检测故障

我不确定哪种方法是最好的，任何指导、链接和信息都会非常有用

json.foreachRDD(new VoidFunction<JavaRDD<String>>() { @Override public void call(JavaRDD<String> rdd) { if(!rdd.isEmpty()){ Dataset<Row> data = spark.read().json(rdd).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"); df.show(false); } } });

Python中文网

有 Java 编程相关的问题?

如何在Java中解析spark流中的复杂JSON数据

共 (1) 个答案

# 1 楼答案