拥有Dataset<Row>
单列 json 字符串:
+--------------------+
| value|
+--------------------+
|{"Context":"00AA0...|
+--------------------+
Json 示例:
{"Context":"00AA00AA","MessageType":"1010","Module":"1200"}
我怎样才能最有效地获得 Dataset<Row>
看起来像这样:
+--------+-----------+------+
| Context|MessageType|Module|
+--------+-----------+------+
|00AA00AA| 1010| 1200|
+--------+-----------+------+
我正在流式处理这些数据,我知道当我从文件中读取数据时,spark 可以自行完成此操作:
spark
.readStream()
.schema(MyPojo.getSchema())
.json("src/myinput")
但现在我正在从 kafka 读取数据,它以另一种形式为我提供数据。 我知道我可以使用一些像 Gson 这样的解析器,但我想让 spark 为我做这件事。
最佳答案
试试这个示例。
public class SparkJSONValueDataset {
public static void main(String[] args) {
SparkSession spark = SparkSession
.builder()
.appName("SparkJSONValueDataset")
.config("spark.sql.warehouse.dir", "/file:C:/temp")
.master("local")
.getOrCreate();
//Prepare data Dataset<Row>
List<String> data = Arrays.asList("{\"Context\":\"00AA00AA\",\"MessageType\":\"1010\",\"Module\":\"1200\"}");
Dataset<Row> df = spark.createDataset(data, Encoders.STRING()).toDF().withColumnRenamed("_1", "value");
df.show();
//convert to Dataset<String> and Read
Dataset<String> df1 = df.as(Encoders.STRING());
Dataset<Row> df2 = spark.read().json(df1.javaRDD());
df2.show();
spark.stop();
}
}
关于java - 将 Json 的 Dataset 列解析为 Dataset<Row>,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/40738190/