PySpark将键值对数组转换为单行结构化数据的技术咨询

我的数据Schema包含headers数组（元素为含key和字符串类型value的结构体）以及id字符串字段，原始数据结构如下：

[{
    key=Username,
    value=
},
{
    key=Email,
    value=
},
{
    key=Id,
    value=
},
{
    key=Organization,
    value=
},
{
    key=Role,
    value=
},
{
    key=Address,
    value=
},
{
    key=Component,
    value=
},
{
    key=Reason,
    value=
},
{
    key=Region,
    value=
}]

对应的Schema：

root
|-- headers: array
|    |-- element: struct
|    |    |-- key: string
|    |    |-- value: string
|-- id: string

我需要将其转换为单行结构，以key作为列名、value作为对应列的数据，目标列包括Username、Email、Id、Organization、Role、Address、Component、Reason、Region。

我已尝试用explode展开键值对为多行，展开后的数据如下：

--------------+-----------------------+----------+
|headers                                                                                      |id        |
+----------------------------------------------------------------------------------------------------+------------------------------------------+-----------------------+----------+
|{Username, }                                                                                    |8239|
|{Email, }                                                                                       |8239|
|{Id, }                                                                                          |8239|
|{Organization, }                                                                                |8239|
|{Role, []}                                                                                      |8239|
|{Address, 9}                                                                            |8239|
|{Component, 9}                                                                             |8239|
|{Reason, 9}                                                                                |8239|
|{Region, 9999}                                                                    |8239|

现寻求实现该转换的正确方法。

以下以Spark为例提供两种高效实现方式：

方法1：Explode + Pivot（适配动态列场景） #

先通过explode展开headers数组，再用pivot将key转成列，最后聚合提取对应value：

Spark SQL 写法 #

SELECT 
  id,
  MAX(CASE WHEN headers.key = 'Username' THEN headers.value END) AS Username,
  MAX(CASE WHEN headers.key = 'Email' THEN headers.value END) AS Email,
  MAX(CASE WHEN headers.key = 'Id' THEN headers.value END) AS Id,
  MAX(CASE WHEN headers.key = 'Organization' THEN headers.value END) AS Organization,
  MAX(CASE WHEN headers.key = 'Role' THEN headers.value END) AS Role,
  MAX(CASE WHEN headers.key = 'Address' THEN headers.value END) AS Address,
  MAX(CASE WHEN headers.key = 'Component' THEN headers.value END) AS Component,
  MAX(CASE WHEN headers.key = 'Reason' THEN headers.value END) AS Reason,
  MAX(CASE WHEN headers.key = 'Region' THEN headers.value END) AS Region
FROM (
  SELECT id, explode(headers) AS headers FROM your_table
) t
GROUP BY id

DataFrame API 写法（Python） #

from pyspark.sql import functions as F

# 展开headers数组
df_exploded = df.select("id", F.explode("headers").alias("header"))
# 透视转列并聚合
df_pivoted = df_exploded.groupBy("id").pivot("header.key").agg(F.first("header.value"))

# 指定目标列顺序（可选）
target_cols = ["Username", "Email", "Id", "Organization", "Role", "Address", "Component", "Reason", "Region"]
df_final = df_pivoted.select("id", *target_cols)

方法2：直接从数组提取指定Key（适配固定列场景） #

无需展开数组，直接用FILTER函数筛选对应key的元素并提取value，效率更高：

Spark SQL 写法 #

SELECT
  id,
  FILTER(headers, x -> x.key = 'Username')[0].value AS Username,
  FILTER(headers, x -> x.key = 'Email')[0].value AS Email,
  FILTER(headers, x -> x.key = 'Id')[0].value AS Id,
  FILTER(headers, x -> x.key = 'Organization')[0].value AS Organization,
  FILTER(headers, x -> x.key = 'Role')[0].value AS Role,
  FILTER(headers, x -> x.key = 'Address')[0].value AS Address,
  FILTER(headers, x -> x.key = 'Component')[0].value AS Component,
  FILTER(headers, x -> x.key = 'Reason')[0].value AS Reason,
  FILTER(headers, x -> x.key = 'Region')[0].value AS Region
FROM your_table

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