时间戳

首页 > 解决方案 > 如何解析包含嵌套 json 形式数据的 kafka 主题？

json - 如何解析包含嵌套 json 形式数据的 kafka 主题？

问题描述

我正在尝试阅读 kafka 主题并将其流式传输到我的接收器。为了读取数据，我编写了以下代码。

json中的主题数据：

{
"HiveData": {
"Tablename": "HiveTablename1",
"Rowcount": "3213423",
"lastupdateddate": "2021-02-24 13:04:14"
},
"HbaseData": [
{
"Tablename": "HbaseTablename1",
"Rowcount": "23543",
"lastupdateddate": "2021-02-23 12:03:11"
}
],
"PostgresData": [
{
"Tablename": "PostgresTablename1",
"Rowcount": "23454345",
"lastupdateddate": "2021-02-23 12:03:11"
}
]
}

以下是我为解析主题而编写的代码：

 def streamData(): DataFrame = {
    val kafkaDF = spark.readStream.format("kafka")
      .option("kafka.bootstrap.servers", "server:port")
      .option("subscribe", "topic_name")
      .load()
    kafkaDF.select(from_json(col("HiveData"), topic_schema).as("HiveData")).selectExpr("HiveData.tablename as table", "HiveData.Rowcount as rowcount", "HiveData.lastupdateddate as lastupdateddate") 
    kafkaDF
 }

但是，如果 json 的格式为：

{"Tablename": "HiveTablename1","Rowcount": "3213423","lastupdateddate": "2021-02-24 13:04:14"}

我想解析 json 并将 HiveData 放入一个单独的数据帧和一个单独的 HBaseData 数据帧和 PostgresData 相同的数据帧。如果 json 数据在一行中，我编写的代码就可以工作。任何人都可以让我知道如何将数据解析为多个数据帧，如果它是本问题开头提到的嵌套格式吗？任何帮助深表感谢。

标签： jsonapache-sparkapache-kafkaspark-streaming

解决方案

尝试添加

选项（“多行”，“真”）

推荐阅读