我使用以下 JSON 結(jié)構(gòu)處理來自 Kafka 的消息：{"unix_time": 1557678233, "category_id": 1000, "ip": "172.10.34.17", "type": "view"}我想打印出我收到的內(nèi)容。這是我已經(jīng)完成的代碼片段：JavaSparkContext sc = createJavaSparkContext();JavaStreamingContext streamingContext =                new JavaStreamingContext(sc, Durations.seconds(BATCH_DURATION_IN_SECONDS));SparkSession sparkSession = SparkSession        .builder()        .config(new SparkConf())        .getOrCreate();Dataset<Row> df = sparkSession        .readStream()        .format("kafka")        .option("kafka.bootstrap.servers", CommonUtils.KAFKA_HOST_PORT)        .option("subscribe", KAFKA_TOPIC)        .load();StreamingQuery query = df.selectExpr("CAST(value AS STRING)")            .select(from_json(new Column("value"), getSchema())).as("data").                    select("data.category_id").writeStream().foreach(new ForeachWriter<Row>() {                @Override                public void process(Row value) {                    System.out.println(value);                }                @Override                public void close(Throwable errorOrNull) {                }                @Override                public boolean open(long partitionId, long version) {                    return true;                }            })            .start();    query.awaitTermination();架構(gòu)方法：private static StructType getSchema() {    return new StructType(new StructField[]{            new StructField(UNIX_TIME, DataTypes.TimestampType, false, Metadata.empty()),            new StructField(CATEGORY_ID, DataTypes.IntegerType, false, Metadata.empty()),            new StructField(IP, DataTypes.StringType, false, Metadata.empty()),            new StructField(TYPE, DataTypes.StringType, false, Metadata.empty()),    });}如何克服這個問題？對此有何建議？