有没有办法在 Pyspark 中动态猜测模式？

Question

我在 Databricks 中有一个表，其中有一列作为字符串字典，如下所示 -

+---+--------------------------------------------------------------------------------------------------------------+
|id |stringDictionary                                                                                              |
+---+--------------------------------------------------------------------------------------------------------------+
|abc|{"col1": "someValue", "col2" : "someValue", "col3" : "someValue"}                                             |
|def|{"col1" : "someValue", "col3": "someValue"}                                                                   |
|mnp|{"col1" : "someValue", "col2" : "someValue", "col3" : "someValue", "col4" : "someValue", "col5" : "someValue"}|
|abc|{"col4" : "someValue", "col5" : "someValue", "col6" : "someValue"}                                            |
+---+--------------------------------------------------------------------------------------------------------------+

现在每个 id 都可以有多个结构，如图所示。

我尝试为其中几个定义 Struct 架构，如下所示 -

from pyspark.sql.types import StructType,StructField, StringType
from pyspark.sql.functions import col,from_json

schema = StructType([ 
    StructField("col1",StringType(),True), 
    StructField("col2",StringType(),True), 
    StructField("col3",StringType(),True), 
    StructField("col4", StringType(), True)
  ])

dfJSON = sparkDF.withColumn("jsonData",from_json(col("stringDictionary"),schema)) \
                   .select("stringDictionary","jsonData.*","*")#.drop("stringDictionary")

display(dfJSON)

但这不是一个好方法，因为每次添加新元素时都不够用，必须手动更改架构。

有没有办法处理所有此类情况或在读取此表或数据帧时猜测结构架构并将其对应的stringDictionary 展平为自己的单独列？

请帮忙。

Answer 1

使用以下代码将 sparkDF 的 stringDictionary 列转换为 MapType -

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

df = (sparkDF.withColumn("cols", 
                          from_json( col("stringDictionary"), 
                                     MapType(StringType(), StringType())
                                    )
                         )
             .drop("stringDictionary")
       )

现在，cols 列需要分解如下 -

df2 = df.select("id", explode("cols").alias("col_columns", "col_rows"))

display(df2)

一旦您将 col_columns 和 col_rows 作为单独的列，我们需要做的就是旋转 col_columns 并使用其对应的 col_rows 聚合它，如下所示 -

df3 = df2.groupBy("id")
         .pivot("col_columns")
         .agg(first("col_rows"))

display(df3)

Answer 2

使用schema_of_json() 动态创建架构，然后使用MergeSchema 进行架构演进。

请注意，在生产环境中，有时 json 有效负载可能会以错误的数据类型发送。此外，模式演化不适用于嵌套结构。