Python中如何基于重复值将多字段JSON加载为指定结构DataFrame

你需要的是同维度下键值对格式的长表转宽表操作，pandas和pyspark都提供了原生内置的pivot行转列接口，不需要编写自定义遍历逻辑即可实现。

Pandas 实现 #

import pandas as pd

# 读取原始数据，以下为示例数据，替换为实际文件读取逻辑即可
raw_data = [
    ('ID_1', 'col1_1', 'col2_1', 'col3_1', 'key1', 'value1', 'col6_1'),
    ('ID_1', 'col1_1', 'col2_1', 'col3_1', 'key2', 'value2', 'col6_1'),
    ('ID_1', 'col1_1', 'col2_1', 'col3_1', 'key3', 'value3', 'col6_1'),
    ('ID_2', 'col1_2', 'col2_2', 'col3_2', 'key1', 'value1', 'col6_2'),
    ('ID_2', 'col1_2', 'col2_2', 'col3_2', 'key2', 'value2', 'col6_2'),
    ('ID_2', 'col1_2', 'col2_2', 'col3_2', 'key3', 'value3', 'col6_2')
]
df = pd.DataFrame(raw_data, columns=['id', 'col1', 'col2', 'col3', 'key_col', 'value_col', 'col6'])

# 行转列核心逻辑
result = df.pivot(
    index=['id', 'col1', 'col2', 'col3', 'col6'],
    columns='key_col',
    values='value_col'
).reset_index()

# 按期望的列顺序重排
result = result[['id', 'col1', 'col2', 'col3', 'key1', 'key2', 'key3', 'col6']]

运行后输出结果和你给出的期望结构完全一致，每个ID对应一行，key1/key2/key3列分别填充对应的value值。

PySpark 实现 #

使用内置pivot接口，大数据量下性能远高于自定义UDF实现：

from pyspark.sql import SparkSession
from pyspark.sql.functions import first

spark = SparkSession.builder.appName("pivot_task").getOrCreate()

# 读取原始数据，以下为示例数据，替换为实际文件读取路径即可
raw_df = spark.createDataFrame([
    ('ID_1', 'col1_1', 'col2_1', 'col3_1', 'key1', 'value1', 'col6_1'),
    ('ID_1', 'col1_1', 'col2_1', 'col3_1', 'key2', 'value2', 'col6_1'),
    ('ID_1', 'col1_1', 'col2_1', 'col3_1', 'key3', 'value3', 'col6_1'),
    ('ID_2', 'col1_2', 'col2_2', 'col3_2', 'key1', 'value1', 'col6_2'),
    ('ID_2', 'col1_2', 'col2_2', 'col3_2', 'key2', 'value2', 'col6_2'),
    ('ID_2', 'col1_2', 'col2_2', 'col3_2', 'key3', 'value3', 'col6_2')
], schema=['id', 'col1', 'col2', 'col3', 'key_col', 'value_col', 'col6'])

# 行转列核心逻辑，提前传入key枚举值可避免全表扫描，大幅提升运行效率
result_df = raw_df.groupBy('id', 'col1', 'col2', 'col3', 'col6')\
    .pivot('key_col', values=['key1', 'key2', 'key3'])\
    .agg(first('value_col'))\
    .select('id', 'col1', 'col2', 'col3', 'key1', 'key2', 'key3', 'col6')

注意事项 #

• 如果同一分组+key组合存在多条重复数据，将聚合函数first()替换为实际需要的逻辑即可，比如max()、collect_list()等
• 如果key的枚举值不固定，可以去掉pivot的values参数，Spark会自动扫描全表识别所有key，仅会额外增加一次全表扫描的开销

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