使用 pyspark 解析 JSON 时嵌套动态模式不起作用

Question

我正在尝试从嵌套的 JSON （具有动态架构） 中提取某些参数并使用 pyspark 生成 spark 数据框。

我的代码在级别 1 (key:value) 上完美运行，但无法为作为嵌套 JSON 一部分的每个 (key:value) 对获取独立列。

JSON 架构示例

注意 - 这不是确切的架构。它只是为了给出模式嵌套性质的想法

{
  "tweet": {
    "text": "RT @author original message"
    "user": {
          "screen_name": "Retweeter"
    },
    "retweeted_status": {
      "text": "original message".
        "user": {         
            "screen_name": "OriginalTweeter"
        },
        "place": {          
        },
        "entities": {          
        },
        "extended_entities": {          
        }
      },     
    },
    "entities": {      
    },
    "extended_entities": {      
    }
  }
}

PySpark 代码

from pyspark.sql.types import StructType, StructField, StringType

schema = StructType([
StructField("text", StringType(), True),
StructField("created_at", StringType(), True),
StructField("retweeted_status", StructType([
StructField("text", StringType(), True),
StructField("created_at", StringType(), True)]))
])

df = spark.read.schema(schema).json("/user/sagarp/NaMo/data/NaMo2019-02-12_00H.json")
df.show()

当前输出 - （带有真实的 JSON 数据）

嵌套 retweet_status JSON 下的所有（键：值）都被压缩为 1 个单个列表。例如[文本，created_at，实体]

+--------------------+--------------------+--------------------+
|                text|          created_at|    retweeted_status|
+--------------------+--------------------+--------------------+
|RT @Hoosier602: @...|Mon Feb 11 19:04:...|[@CLeroyjnr @Gabr...|
|RT @EgSophie: Oh ...|Mon Feb 11 19:04:...|[Oh cool so do yo...|
|RT @JacobAWohl: @...|Mon Feb 11 19:04:...|[@realDonaldTrump...|

预期输出

我希望每个键都有独立的列。另外，请注意您已经有一个同名的父级密钥 text。您将如何处理此类情况？

理想情况下，我想要“text”、“entities”、“retweet_status_text”、“retweet_status_entities”等列

Answer 1

您的架构未正确映射...如果您想手动构建架构，请参阅这些帖子（如果数据不更改，建议这样做）：

PySpark: How to Update Nested Columns?

https://docs.databricks.com/_static/notebooks/complex-nested-structured.html

另外，如果您的 JSON 是多行的（如您的示例），那么您可以...

1. 通过多行选项读取 json 以让 Spark 推断架构
2. 然后保存嵌套架构
3. 然后使用正确的架构映射读回数据以避免触发 Spark 作业

! cat nested.json

[
    {"string":"string1","int":1,"array":[1,2,3],"dict": {"key": "value1"}},
    {"string":"string2","int":2,"array":[2,4,6],"dict": {"key": "value2"}},
    {
        "string": "string3",
        "int": 3,
        "array": [
            3,
            6,
            9
        ],
        "dict": {
            "key": "value3",
            "extra_key": "extra_value3"
        }
    }
]

getSchema = spark.read.option("multiline", "true").json("nested.json")

extractSchema = getSchema.schema
print(extractSchema)
StructType(List(StructField(array,ArrayType(LongType,true),true),StructField(dict,StructType(List(StructField(extra_key,StringType,true),StructField(key,StringType,true))),true),StructField(int,LongType,true),StructField(string,StringType,true)))

loadJson = spark.read.option("multiline", "true").schema(extractSchema ).json("nested.json")

loadJson.printSchema()
root
 |-- array: array (nullable = true)
 |    |-- element: long (containsNull = true)
 |-- dict: struct (nullable = true)
 |    |-- extra_key: string (nullable = true)
 |    |-- key: string (nullable = true)
 |-- int: long (nullable = true)
 |-- string: string (nullable = true)

loadJson.show(truncate=False)
+---------+----------------------+---+-------+
|array    |dict                  |int|string |
+---------+----------------------+---+-------+
|[1, 2, 3]|[, value1]            |1  |string1|
|[2, 4, 6]|[, value2]            |2  |string2|
|[3, 6, 9]|[extra_value3, value3]|3  |string3|
+---------+----------------------+---+-------+

使用正确的映射加载数据后，您就可以开始通过嵌套列的“点”表示法和“展开”以展平数组等转换为规范化架构。

loadJson\
.selectExpr("dict.key as key", "dict.extra_key as extra_key").show()

+------+------------+
|   key|   extra_key|
+------+------------+
|value1|        null|
|value2|        null|
|value3|extra_value3|
+------+------------+