PySpark多列透视需求：按Job #分组转换故障码及状态列

修正后的代码示例 #

首先确保导入PySpark函数库：

from pyspark.sql import functions as F

如果需要按业务逻辑（比如取最新状态）处理重复记录，先对数据排序再透视：

# 若有时间字段，按Job #和时间倒序排序，保证取到最新的trouble_status
sorted_df = mydf.orderBy("Job #", F.col("record_time").desc())

# 生成带前缀的列名，执行透视操作
trouble_df = sorted_df.withColumn(
    'combcol', 
    F.concat(F.lit('trouble_code_'), F.col('trouble_code'))
).groupBy('Job #').pivot('combcol').agg(F.first('trouble_status'))

常见问题排查 #

1. 重复记录导致取值不符合预期
   同一Job #下相同trouble_code如果有多条记录，F.first()会取数据集中的第一条。如果未提前排序，取到的可能不是你需要的状态。建议根据业务规则（如时间、优先级）排序后再执行透视。

2. 空值生成多余列
   如果源数据中trouble_code存在null值，会生成trouble_code_null列。若不需要这类列，先过滤空值：

   mydf = mydf.filter(F.col('trouble_code').isNotNull())
   

3. 透视列过多影响性能
   若trouble_code取值范围大，pivot会自动生成大量列，可提前指定需要透视的列名列表来优化：

   # 获取所有唯一的trouble_code并生成前缀列名
   target_cols = [f"trouble_code_{code}" for code in mydf.select("trouble_code").distinct().rdd.flatMap(lambda x: x).collect()]
   
   trouble_df = sorted_df.withColumn(
       'combcol', 
       F.concat(F.lit('trouble_code_'), F.col('trouble_code'))
   ).groupBy('Job #').pivot('combcol', target_cols).agg(F.first('trouble_status'))
   

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