如何在PySpark中按列分组透视列，同时保持item_type升序

你的问题核心是全局排序无法保证分组后collect_list的元素顺序——Spark的分布式计算特性导致groupBy时，同组数据可能分散在不同分区，collect_list会按分区内的存储顺序收集元素，和全局orderBy的结果无关。

正确解决方案：用Window函数生成分组内有序索引，再Pivot转宽表 #

通过给每个(date, item_id)分组内的行按item_type升序分配行号，再利用pivot将行转成列，既能保证顺序，又能自动填充不足5个的null值。

修改后的完整代码：

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, IntegerType
from pyspark.sql.window import Window
from pyspark.sql.functions import row_number, first, col

# 创建SparkSession
spark = SparkSession.builder.appName("example").getOrCreate()

# 定义Schema和测试数据
schema = StructType([
    StructField("date", StringType(), True),
    StructField("item_id", StringType(), True),
    StructField("item_type", StringType(), True),
    StructField("item_vol", IntegerType(), True)
])

data = [
    ('2019-01-01', 'item3', 'aa', 1),
    ('2019-01-01', 'item3', 'bb', 2),
    ('2019-01-01', 'item67', 'cc', 4),
    ('2019-01-01', 'item67', 'dd', None),
    ('2019-01-01', 'item68', 'gas', 9),
]

new_df = spark.createDataFrame(data, schema)

# 1. 给每个(date, item_id)分组内的行按item_type升序分配行号
w = Window.partitionBy("date", "item_id").orderBy("item_type")
ranked_df = new_df.withColumn("rn", row_number().over(w))

# 2. 按行号pivot，生成对应列，不足5个的自动补null
pivoted_df = ranked_df.groupBy("date", "item_id") \
    .pivot("rn", range(1, 6))  # 指定pivot的行号范围1-5，确保生成5列
    .agg(
        first("item_type").alias("item_type"),
        first("item_vol").alias("item_vol")
    )

# 3. 重命名列，符合item_type_1、item_vol_1的格式
final_df = pivoted_df.select(
    "date", "item_id",
    *[col(f"{i}_item_type").alias(f"item_type_{i}") for i in range(1, 6)],
    *[col(f"{i}_item_vol").alias(f"item_vol_{i}") for i in range(1, 6)]
)

# 查看结果
final_df.show(10, False)

关键逻辑说明 #

• Window函数：partitionBy("date", "item_id")确保只在同一日期和商品ID内排序，orderBy("item_type")保证item_type升序，row_number()给每个分组内的行分配唯一序号（1、2...）。
• Pivot指定范围：pivot("rn", range(1,6))强制生成1到5的列，即使某个分组不足5条数据，对应列也会填充null。
• agg取first值：因为每个行号在分组内唯一，first()能准确获取对应行的item_type和item_vol。

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