spark为您提供了将输入json读取到数据集/Dataframe的api。
如果您正在从文件中读取json。你可以使用
SparkSession read().json() 方法

SparkSession spark = SparkSession.builder().appName("Test App").master("local[*]")
                .getOrCreate();

Dataset<Row> inputfileDataset = spark.read().option("multiLine", true).option("mode", "PERMISSIVE")
            .json("C:\path-to-file\inputfile.json");

如果你在读Kafka的作品。您可以通过将每个rdd转换为 List 的 String 然后将它们转换为数据集/Dataframe

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

JavaDStream<String> inputJsonStream = directKafkaStream.map(rdd -> {
    return rdd._2;
});

inputJsonStream.foreachRDD(inputRDD -> {
    SparkSession spark = JavaSparkSessionSingleton.getInstance(inputRDD.context().getConf());
    List<String> strings = inputRDD.collect();
    strings.forEach(x -> {
        Dataset<Row> inputDataset = spark.read().option("multiLine",true).option("mode", "PERMISSIVE").json(inputRDD);
        inputDataset.printSchema();
});

对于查询数据集/Dataframe，可以使用 Select() 函数，然后将其转换为所需的数据类型。