将嵌套字典键值转换为 pyspark 数据框

Posted 2023-04-15

【中文标题】将嵌套字典键值转换为 pyspark 数据框

【英文标题】：Transform nested dictionary key values to pyspark dataframe

【发布时间】：2020-07-20 22:15:08

【问题描述】：

我有一个如下所示的 Pyspark 数据框：

我想提取“dic”列中的那些嵌套字典并将它们转换为 PySpark 数据框。像这样：

请告诉我如何实现这一目标。

谢谢！

【参考方案1】：

from pyspark.sql import functions as F

df.show() #sample dataframe

+---------+----------------------------------------------------------------------------------------------------------+
|timestmap|dic                                                                                                       |
+---------+----------------------------------------------------------------------------------------------------------+
|timestamp|"Name":"David","Age":"25","Location":"New York","Height":"170","fields":"Color":"Blue","Shape":"round"|
+---------+----------------------------------------------------------------------------------------------------------+

对于 Spark2.4+，您可以使用 from_json 和 schema_of_json。

schema=df.select(F.schema_of_json(df.select("dic").first()[0])).first()[0]


df.withColumn("dic", F.from_json("dic", schema))\
  .selectExpr("dic.*").selectExpr("*","fields.*").drop("fields").show()

#+---+------+--------+-----+-----+-----+
#|Age|Height|Location| Name|Color|Shape|
#+---+------+--------+-----+-----+-----+
#| 25|   170|New York|David| Blue|round|
#+---+------+--------+-----+-----+-----+

如果您没有 spark2.4，您也可以使用 rdd 方式和 read.json。 df to rdd 转换会影响性能。

df1 = spark.read.json(df.rdd.map(lambda r: r.dic))\
   
df1.select(*[x for x in df1.columns if x!='fields'], F.col("fields.*")).show()

#+---+------+--------+-----+-----+-----+
#|Age|Height|Location| Name|Color|Shape|
#+---+------+--------+-----+-----+-----+
#| 25|   170|New York|David| Blue|round|
#+---+------+--------+-----+-----+-----+

【讨论】：

谢谢！如何在无需手动构建的情况下获取架构？ （我上面提供的json只是示例数据。我的真实数据中至少有1000个字段）。

您可以使用 schema_of_json 函数为您执行此操作，如我上面编辑的答案所示。唯一的事情是 schema_of_json 只接受一个字符串文字，所以我们必须收集一个行字符串并将其传递给函数。这里的问题是您正在收集的行应该包含所有字段

感谢您的帮助。我刚刚意识到我的数据实际上有 3 层嵌套。使用您的解决方案我只能获得第二个级别我怎样才能从这里接近第三个级别？ （我已经编辑了我的问题）