求助：基于PySpark按Code聚合Value并展开为多列的实现方案

核心思路 #

要处理这种多组code-value列的聚合需求，关键是先将宽表转长表（unpivot），统一聚合所有code对应的value总和，再长表转宽表（pivot）生成目标列，最后关联原表并计算总计。这种方式能兼容最多100组code-value列的动态场景，无需硬编码列名。

分步实现代码 #

假设你的原始DataFrame名为df，以下是完整实现逻辑：

1. 动态识别Code-Value列对

自动匹配所有code_*列及其对应的value_*列（忽略大小写差异，比如Value_1或value_3都能匹配）：

from pyspark.sql import functions as F

# 按数字后缀排序code列，确保顺序正确
code_cols = sorted(
    [col for col in df.columns if col.startswith('code_')],
    key=lambda x: int(x.split('_')[1])
)

# 匹配对应后缀的value列
value_cols = []
for code_col in code_cols:
    idx = code_col.split('_')[1]
    # 查找对应数字后缀的value列（忽略大小写）
    value_col = next(
        (col for col in df.columns if col.lower() == f'value_{idx}'),
        None
    )
    if value_col:
        value_cols.append(value_col)

2. 宽表转长表（Unpivot）

将多组code-value列转换成每行一个code和对应value的长表结构：

# 构造stack表达式，动态适配code-value列的数量
stack_count = len(code_cols)
stack_args = ", ".join([f"'{c}', {v}" for c, v in zip(code_cols, value_cols)])
stack_expr = f"stack({stack_count}, {stack_args})"

# 生成长表，保留id1用于后续关联
unpivoted_df = df.select("id1", F.expr(stack_expr).alias("code", "value"))

3. 聚合每个Code的总和

按code分组，计算全局的value总和：

aggregated_df = unpivoted_df.groupBy("code").agg(F.sum("value").alias("sum_val"))

4. 长表转宽表（Pivot）

将聚合结果转回宽表，列名格式改为sum_<code>：

# pivot生成以code为列名的宽表，再重命名列
pivoted_df = aggregated_df.groupBy().pivot("code").agg(F.first("sum_val"))
pivoted_df = pivoted_df.select([F.col(col).alias(f"sum_{col}") for col in pivoted_df.columns])

5. 关联原表并计算总计

将原表的id1和item列与聚合后的宽表做交叉关联（因为聚合结果是全局唯一的一行），再计算所有sum列的总和得到Total：

# 关联原表的核心字段
result_df = df.select("id1", "item").crossJoin(pivoted_df)

# 计算Total列
sum_columns = [col for col in result_df.columns if col.startswith("sum_")]
result_df = result_df.withColumn("Total", F.sum(*sum_columns))

6. 查看结果

result_df.show()

关键说明 #

• 动态适配：代码会自动识别所有code_*和对应的value_*列，不管有多少组（100组也能处理），无需修改代码。
• 全局聚合：示例中是对整个表的code值求和，若需按id1分组聚合，只需在第2步的unpivoted_df中保留id1，并在第3步的groupBy中加入id1，后续pivot和关联时也按id1关联即可。
• 列名兼容：处理了Value_*和value_*的大小写差异，确保不会漏匹配列。

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