用于动态关键字段的 Spark JSON 模式？答案

【问题标题】：Spark JSON Schema for dynamic key fields?用于动态关键字段的 Spark JSON 模式？
【发布时间】：2019-07-21 05:43:58
【问题描述】：

我使用 from_json() 方法从 kafka 接收 JSON 数据。它需要我的架构。我的 JSON 结构是这样的；

{
    "Items": {
        "key1": [
            {
                "id": "",
                "name": "",
                "val": ""
            }
        ],
        "key2": [
            {
                "id": "",
                "name": "",
                "val": ""
            }
        ],
        "key3": [
            {
                "id": "",
                "name": "",
                "val": ""
            }
        ]
    }
}

在这种状态下； key1, key2, key3 字段未知。所以，他们是dynamic。这些字段名称可能是"abc", "def" 等。如何在 Spark Structured Streaming 中为此 JSON 定义 json 架构？

编辑： 比如另一个json；

{
        "Items": {
            "stack": [
                {
                    "id": "",
                    "name": "",
                    "val": ""
                }
            ],
            "over": [
                {
                    "id": "",
                    "name": "",
                    "val": ""
                }
            ],
            "flow": [
                {
                    "id": "",
                    "name": "",
                    "val": ""
                }
            ]
        }
    }

【问题讨论】：

标签： json apache-spark apache-spark-sql spark-structured-streaming

【解决方案1】：

您需要定义您的架构来解析此类 json 文件。您的架构应如下所示。

val valSchema = new StructType()
  .add("id", StringType)
  .add("name", StringType)
  .add("val", StringType)

val valArrSchema = new ArrayType(valSchema, true)
val mapSchema = new MapType(StringType, valArrSchema, true)
val jsonSchema =  new StructType().add("Items", mapSchema)

您可以使用 from_json 方法解析 json。

val testDF = df.withColumn("json", from_json(col("value"),jsonSchema ))

【讨论】：

【解决方案2】：

key1、key2 和 key3 不是归档文件。它们是关键的 av 值！这里的字段是id、name、value 和key，key 的值可以是动态的，没关系，这里没有复杂性。

【讨论】：