如何在PySpark中解析嵌套JSON并展开为指定结构的扁平DataFrame？

需求 #

将给定的嵌套JSON解析为指定结构的扁平多列DataFrame。

给定嵌套JSON示例 #

{
  "ID": "abc123",
  "device": "mobile",
  "Ads": [
    {
      "placement": "topright",
      "Adlist": [
        {
          "name": "ad1",
          "subtype": "placeholder1",
          "category": "socialmedia"
        },
        {
          "name": "ad2",
          "subtype": "placeholder2",
          "category": "media"
        },
        {
          "name": "ad3",
          "subtype": "placeholder3",
          "category": "printingpress"
        }
      ]
    },
    {
      "Placement": "bottomleft",
      "Adlist": [
        {
          "name": "ad4",
          "subtype": "placeholder4",
          "category": "socialmedia"
        },
        {
          "name": "ad5",
          "subtype": "placeholder5",
          "category": "media"
        },
        {
          "name": "ad6",
          "subtype": "placeholder6",
          "category": "printingpress"
        }
      ]
    }
  ]
}

已尝试代码 #

from pyspark import SparkContext
sc = SparkContext.getOrCreate(SparkConf().setMaster("local[*]"))

rdd = sc.parallelize([sample_json])
df = spark.read.option('multiline', "true").json(rdd)

df.printSchema()

当前DataFrame Schema #

root
 |-- Ads: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- Adlist: array (nullable = true)
 |    |    |    |-- element: struct (containsNull = true)
 |    |    |    |    |-- category: string (nullable = true)
 |    |    |    |    |-- name: string (nullable = true)
 |    |    |    |    |-- subtype: string (nullable = true)
 |    |    |-- placement: string (nullable = true)
 |-- ID: string (nullable = true)
 |-- device: string (nullable = true)

当前DataFrame内容 #

+--------------------+------+------+
|                 Ads|    ID|device|
+--------------------+------+------+
|[{[{socialmedia, ...|abc123|mobile|
+--------------------+------+------+

期望DataFrame结构 #

Id    | device | placement | name  | subtype      | category      |
-------------------------------------------------------------------
abc123| mobile | topright  | ad1   | placeholder1 | socialmedia   |
abc123| mobile | topright  | ad2   | placeholder2 | media         |
abc123| mobile | topright  | ad3   | placeholder3 | printingpress |
abc123| mobile |bottomleft | ad4   | placeholder4 | socialmedia   |
abc123| mobile |bottomleft | ad5   | placeholder5 | media         |
abc123| mobile |bottomleft | ad6   | placeholder6 | printingpress |

解决方案 #

通过两次explode操作展开嵌套数组，同时处理JSON中placement字段大小写不一致的问题：

完整代码 #

from pyspark.sql import SparkSession
from pyspark.sql.functions import explode, col, coalesce

# 初始化SparkSession（PySpark标准初始化方式）
spark = SparkSession.builder.master("local[*]").appName("JsonFlatten").getOrCreate()

# 定义JSON字符串
sample_json = """
{
  "ID": "abc123",
  "device": "mobile",
  "Ads": [
    {
      "placement": "topright",
      "Adlist": [
        {
          "name": "ad1",
          "subtype": "placeholder1",
          "category": "socialmedia"
        },
        {
          "name": "ad2",
          "subtype": "placeholder2",
          "category": "media"
        },
        {
          "name": "ad3",
          "subtype": "placeholder3",
          "category": "printingpress"
        }
      ]
    },
    {
      "Placement": "bottomleft",
      "Adlist": [
        {
          "name": "ad4",
          "subtype": "placeholder4",
          "category": "socialmedia"
        },
        {
          "name": "ad5",
          "subtype": "placeholder5",
          "category": "media"
        },
        {
          "name": "ad6",
          "subtype": "placeholder6",
          "category": "printingpress"
        }
      ]
    }
  ]
}
"""

# 读取JSON数据
rdd = spark.sparkContext.parallelize([sample_json])
df = spark.read.option('multiline', "true").json(rdd)

# 第一步：展开Ads数组，拆分每个广告位置为单独行
df_explode_ads = df.select("ID", "device", explode("Ads").alias("ad_struct"))

# 第二步：提取placement字段（处理大小写不一致）和Adlist数组
df_placement = df_explode_ads.select(
    "ID",
    "device",
    coalesce(col("ad_struct.placement"), col("ad_struct.Placement")).alias("placement"),
    "ad_struct.Adlist"
)

# 第三步：展开Adlist数组，提取每个广告的详情字段
df_final = df_placement.select(
    col("ID").alias("Id"),
    "device",
    "placement",
    explode("Adlist").alias("ad_details")
).select(
    "Id",
    "device",
    "placement",
    "ad_details.name",
    "ad_details.subtype",
    "ad_details.category"
)

# 查看结果
df_final.show()

代码说明 #

1. SparkSession初始化：替代直接操作SparkContext，符合PySpark最佳实践。
2. 第一次展开：用explode拆分Ads数组，将每个广告位置转为单独行，保留顶层的ID和device字段。
3. 处理字段大小写：用coalesce函数统一placement字段，优先取小写值，为空则取大写值。
4. 第二次展开：再次用explode拆分每个位置下的Adlist数组，提取广告的具体属性。
5. 字段重命名：将原ID改为Id，匹配期望的结构。

执行结果 #

运行代码后得到的DataFrame与期望结构完全一致：

+------+------+----------+----+-------------+-------------+
|    Id|device| placement|name|      subtype|     category|
+------+------+----------+----+-------------+-------------+
|abc123|mobile|  topright| ad1|placeholder1|  socialmedia|
|abc123|mobile|  topright| ad2|placeholder2|        media|
|abc123|mobile|  topright| ad3|placeholder3|printingpress|
|abc123|mobile| bottomleft| ad4|placeholder4|  socialmedia|
|abc123|mobile| bottomleft| ad5|placeholder5|        media|
|abc123|mobile| bottomleft| ad6|placeholder6|printingpress|
+------+------+----------+----+-------------+-------------+

内容的提问来源于stack exchange，提问作者Gingerbread