使用不同级别的两个列表分解嵌套 JSON 文件

Question

我有一个 json 文件，布局如下。

{
  "$schema": "schema",
  "records": [
    {
      "mode": "1",      
      "fields": [
        {
          "name": "id",
          "value": "111"
        },
        {
          "name": "name_1a",
          "value": "value_1a"
        },
        {
          "name": "name_1b",
          "value": "value_1b"
        }
      ]
    },
    {
      "mode": "2",
      "fields": [
        {
          "name": "id",
          "value": "222"
        },
        {
          "name": "name_2a",
          "value": "value_2a"
        },
        {
          "name": "name_2b",
          "value": "value_2b"
        }
      ]
    }
  ],
  "format": "json"
}

我正在尝试将其读入 pyspark 数据框以实现以下目标：

id	name	value
111	mode	1
111	name_1a	value_1a
111	name_1b	value_1b
222	mode	2
222	name_2a	value_2a
222	name_2b	value_2b

我已经能够通过 select 适当的键分解出各个元素，例如：

data_fields= data_fields.select('records.mode', 'records.fields')
data_fields = data_fields.select(explode("mode"))

在字段的情况下，我分解了两次以找到较低级别的名称值对。要提取 ids ("111,222") 我 select 从列表中提取该元素，然后使用以下方法再次展开字段：

data_fields = data_fields.withColumn('id', col('fields')[0].value)

但这导致了问题。我不确定如何可靠地将“模式”记录附加到与字段记录相同的格式。

我该怎么做？

Answer 1

内联结构数组后，您可以使用 when 表达式进行一些转换：

from pyspark.sql import functions as F, Window

result = (data_fields.selectExpr("inline(records)")
         .selectExpr("mode", "inline(fields)")
         .withColumn("id", F.max(F.when(F.col("name") == "id", F.col("value"))).over(Window.partitionBy("mode")))
         .withColumn("value", F.when(F.col("name") == "id", F.col("mode")).otherwise(F.col("value")))
         .withColumn("name", F.when(F.col("name") == "id", "mode").otherwise(F.col("name")))
         .drop("mode")
         )

result.show()
#+-------+--------+---+
#|   name|   value| id|
#+-------+--------+---+
#|   mode|       1|111|
#|name_1a|value_1a|111|
#|name_1b|value_1b|111|
#|   mode|       2|222|
#|name_2a|value_2a|222|
#|name_2b|value_2b|222|
#+-------+--------+---+

使用不同级别的两个列表分解嵌套 JSON 文件

Explode nested JSON file with two lists at different levels

python

json

apache-spark

pyspark

apache-spark-sql