PySpark按列名实现数据透视及缺失值填充求助

核心思路 #

通过拆分列名提取提供商（provider）和字段名，使用PySpark的stack函数实现宽表转窄表，同时处理单个提供商独有的字段，确保其他提供商对应字段值为null。

实现代码 #

修改原JobBase类的execute方法，添加转换逻辑：

import sys,os
import concurrent.futures
from concurrent.futures import *
import boto3
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from pyspark.context import SparkConf
from awsglue.context import GlueContext
from awsglue.job import Job
from awsglue.dynamicframe import DynamicFrame

from pyspark.sql import DataFrame
from pyspark.sql.functions import upper


###############################

class JobBase(object):
    spark=None
    def __start_spark_glue_context(self):
        conf = SparkConf().setAppName("python_thread")
        self.sc = SparkContext(conf=conf)
        self.glueContext = GlueContext(self.sc)
        self.spark = self.glueContext.spark_session

    def execute(self):
        self.__start_spark_glue_context()
        new_dict={}
        print('hello')
        
        # 示例输入数据（包含单个提供商独有的字段）
        d = [{"v1_ind": 'A', "v1_rev": 23,"v2_ind": 'b', "v2_rev": 44, "v1_extra": "only_v1"}]

        df = self.spark.createDataFrame(data=d)
        print("原始数据:")
        df.show()        
        
        # 1. 拆分列名，提取provider和字段名
        column_map = {}
        for col in df.columns:
            if '_' in col:
                provider, field = col.split('_', 1)
                if provider not in column_map:
                    column_map[provider] = []
                column_map[provider].append(field)
        
        # 2. 获取所有唯一字段名，统一字段集合
        all_fields = set()
        for fields in column_map.values():
            all_fields.update(fields)
        all_fields = sorted(all_fields)
        
        # 3. 构造stack表达式，处理每个provider的字段，缺失字段填充null
        stack_clauses = []
        for provider in column_map:
            field_values = []
            for field in all_fields:
                col_name = f"{provider}_{field}" if f"{provider}_{field}" in df.columns else "null"
                field_values.append(col_name)
            # 拼接单个provider的stack子句
            clause = f"'{upper(provider)}', {', '.join(field_values)}"
            stack_clauses.append(clause)
        
        # 4. 生成完整的selectExpr语句
        num_providers = len(column_map)
        num_fields = len(all_fields)
        stack_expr = f"stack({num_providers}, {', '.join(stack_clauses)}) as (provider, {', '.join([upper(f) for f in all_fields])})"
        
        # 5. 执行转换
        result_df = df.selectExpr(stack_expr)
        
        print("转换后数据:")
        result_df.show()        
        
def main():
    job = JobBase()
    job.execute() 
    

if __name__ == '__main__':
    main()

代码说明 #

• 列名拆分：遍历所有列，按下划线拆分出provider和字段名，存储为字典结构，键为provider，值为对应字段列表。
• 统一字段集合：收集所有出现过的字段名，确保每个provider的转换记录都包含全量字段，缺失字段用null填充。
• 构造stack表达式：利用stack(n, ...)函数将n组provider数据转成多行记录，每组记录包含provider名称和对应字段值。
• 大小写处理：通过upper函数将provider和字段名转为大写，匹配示例输出格式。

运行结果 #

原始数据：

+------+------+------+------+--------+
|v1_ind|v1_rev|v2_ind|v2_rev|v1_extra|
+------+------+------+------+--------+
|     A|    23|     b|    44| only_v1|
+------+------+------+------+--------+

转换后数据：

+--------+---+---+-----+
|provider|IND|REV|EXTRA|
+--------+---+---+-----+
|      V1|  A| 23|only_v1|
|      V2|  b| 44| null|
+--------+---+---+-----+

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