读取 Pyspark Struct Json 列非必需元素答案

【问题标题】：Read Pyspark Struct Json Column Non Required elements读取 Pyspark Struct Json 列非必需元素
【发布时间】：2020-10-29 07:16:51
【问题描述】：

我有一个 parquet 文件，其中一列是一个存储 json 的结构字段。结构如下图。

originator: struct (nullable = true)
    |-- originatorDetail: struct (nullable = true)
    |    |-- applicationDeployedId: string (nullable = true)
    |    |-- applicationDeployedNameVersion: string (nullable = true)
    |    |-- applicationNameVersion: string (nullable = true)
    |    |-- cloudHost: string (nullable = true)
    |    |-- cloudRegion: string (nullable = true)
    |    |-- cloudStack: string (nullable = true)
    |    |-- version: string (nullable = true)
    |-- Orversion: string (nullable = true)

json中只有version字段是必须的，其他都是非必填字段。所以有些记录可能只有 2 个元素并且仍然有效。

假设我想读取 CloudHost 字段。我可以将其解读为 originator.originatorDetail.cloudHost。但对于不存在此非必填字段的记录。它会失败，因为该元素不存在。对于不使用 udf 的值不存在的记录，有什么方法可以将这些非必需值读取为 null。

一些例子

originator": {
    "originatorDetail": {
      "applicationDeployedId": "PSLV",
      "cloudRegion": "Mangal",
      "cloudHost": "Petrol",
      "applicationNameVersion": "CRDI",
      "applicationDeployedNameVersion": "Tuna",
      "cloudStack": "DEV",
      "version": "1.1.0"
    },
    Orversion": "version.1"
  }
  -------------
 originator": {
    "originatorDetail": {
      "version": "1.1.0"
    },
    Orversion": "version.1"
  }

需要的输出

applicationDeployedId applicationDeployedNameVersion  applicationNameVersion cloudHost cloudRegion cloudStack version  Orversion
 PSLV                   Tuna                            CRDI                   Petrol    Mangal       DEV       1.1.0    version.1
                                                                                                                1.1.0    version.1

【问题讨论】：

它可以帮助你stackoverflow.com/questions/58171816/…

标签： json apache-spark pyspark

【解决方案1】：

使用 Spark-2.4+

中的 from_json 函数

读取镶木地板数据，然后通过传递与您的 json 列匹配的 schema 来使用 from_json。

Spark 将读取匹配的数据并添加空值的非匹配字段。

Example:

df.show(10,False)
#+---+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
#|id |json_data                                                                                                                                                                                                                                                      #|
#+---+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
#|1  |{"originator": {"originatorDetail": {"applicationDeployedId": "PSLV","cloudRegion": "Mangal","cloudHost": "Petrol","applicationNameVersion": "CRDI","applicationDeployedNameVersion": "Tuna","cloudStack": "DEV","version": "1.1.0"},"Orversion": "version.1"}}|
#|2  |{"originator": {    "originatorDetail": {      "version": "1.1.0"    },    "Orversion": "version.1"}}                                                                                                                                                          |
#+---+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
schema=StructType([StructField("originator",StructType([StructField("Orversion",StringType(),True),
            StructField("originatorDetail",StructType([StructField("applicationDeployedId",StringType(),True),
            StructField("applicationDeployedNameVersion",StringType(),True),
            StructField("applicationNameVersion",StringType(),True),
            StructField("cloudHost",StringType(),True),
            StructField("cloudRegion",StringType(),True),
            StructField("cloudStack",StringType(),True),
            StructField("version",StringType(),True)]),True)]),True)])

from pyspark.sql.functions import *
from pyspark.sql.types import *

#then read the json_data column using from_json function
df.withColumn("json_converted",from_json(col("json_data"),schema)).select("id","json_converted").show(10,False)
#+---+--------------------------------------------------------+
#|id |json_converted                                          |
#+---+--------------------------------------------------------+
#|1  |[[version.1, [PSLV, Tuna,, Petrol, Mangal, DEV, 1.1.0]]]|
#|2  |[[version.1, [,,,,,, 1.1.0]]]                           |
#+---+--------------------------------------------------------+

df.withColumn("json_converted",from_json(col("json_data"),schema)).select("id","json_converted").printSchema()
#root
# |-- id: long (nullable = true)
# |-- json_converted: struct (nullable = true)
# |    |-- originator: struct (nullable = true)
# |    |    |-- Orversion: string (nullable = true)
# |    |    |-- originatorDetail: struct (nullable = true)
# |    |    |    |-- applicationDeployedId: string (nullable = true)
# |    |    |    |-- applicationDeployedNameVersion: string (nullable = true)
# |    |    |    |-- applicationNameVersi: string (nullable = true)
# |    |    |    |-- cloudHost: string (nullable = true)
# |    |    |    |-- cloudRegion: string (nullable = true)
# |    |    |    |-- cloudStack: string (nullable = true)
# |    |    |    |-- version: string (nullable = true)

#even though we don't have all fields from id=2 still we added fields
df.withColumn("json_converted",from_json(col("json_data"),schema)).select("json_converted.originator.originatorDetail.applicationDeployedId").show(10,False)
#+---------------------+
#|applicationDeployedId|
#+---------------------+
#|PSLV                 |
#|null                 |
#+---------------------+

【讨论】：