apache-spark - 在 PySpark 中处理嵌套的 JSON

Question

我正在将一个 JSON 文件加载到 PySpark 中:

df = spark.read.json("20220824211022.json")
df.show()

+--------------------+--------------------+--------------------+
|                data|            includes|                meta|
+--------------------+--------------------+--------------------+
|[{961778216070344...|{[{2018-02-09T01:...|{1562543391161741...|
+--------------------+--------------------+--------------------+

我感兴趣的两列是data 和includes。对于 data，我运行了以下命令:

df2 = df.withColumn("data", F.explode(F.col("data"))).select("data.*")
df2.show(2)

+-------------------+--------------------+-------------------+--------------+--------------------+
|          author_id|          created_at|                 id|public_metrics|                text|
+-------------------+--------------------+-------------------+--------------+--------------------+
| 961778216070344705|2022-08-24T20:52:...|1562543391161741312|  {0, 0, 0, 2}|With Kaskada, you...|
|1275784834321768451|2022-08-24T20:47:...|1562542031284555777|  {2, 0, 0, 0}|Below is a protot...|
+-------------------+--------------------+-------------------+--------------+--------------------+

这是我可以使用的东西。但是，我不能对 includes 列执行相同的操作，因为它有包含 [] 的 {}。

我有办法使用 PySpark 来处理这个问题吗？

编辑:

如果您查看 JSON 文件中的 includes 部分，它看起来像:

"includes": {"users": [{"id": "893899303" .... }, ...]},

理想情况下，在我的问题的第一个表中，我希望 includes 成为 users，或者至少能够深入到 users

Answer 1

由于您的 includes 列是一个 MapType 且 key value = "users"，您可以使用 .getItem() 通过key获取数组，即:

df3 = df.withColumn("includes", F.explode(F.col("includes").getItem("users"))).select("includes.*")