如何在PySpark中将用户ID作为行、查询作为列重构DataFrame

要实现这种行列转换，需要先为每个用户的查询按顺序生成序号，再通过pivot函数完成转置，具体步骤如下：

1. 导入必要函数 #

需要用到窗口函数来为每个用户的查询分配序号：

from pyspark.sql import Window
from pyspark.sql.functions import row_number, col, first

2. 为每个用户的查询添加序号 #

通过Window.partitionBy("AnonID")按用户分组，再根据数据的原有顺序（或时间字段，如果有的话）生成查询序号：

# 定义窗口：按AnonID分组，若有时间字段建议替换为时间字段排序，比如orderBy("QueryTime")
window_spec = Window.partitionBy("AnonID").orderBy(col("AnonID"))

# 添加序号列，命名为query_num
df_with_num = df.select("AnonID", "Query") \
                .withColumn("query_num", row_number().over(window_spec))

如果数据包含查询时间字段，用时间排序能保证序号与实际查询顺序一致，示例如下：

# 假设存在查询时间字段QueryTime，替换为你的实际字段名
window_spec = Window.partitionBy("AnonID").orderBy("QueryTime")

3. 使用pivot完成转置 #

按用户ID分组，以序号为列名，查询内容为值进行转置：

# 分组转置，用first()取每个序号对应的查询内容
result_df = df_with_num.groupBy("AnonID") \
                       .pivot("query_num") \
                       .agg(first("Query"))

# 可选：将列名重命名为更直观的格式，比如Query1、Query2...
for col_name in result_df.columns:
    if col_name != "AnonID":
        result_df = result_df.withColumnRenamed(col_name, f"Query{col_name}")

# 展示结果
result_df.show()

关键说明 #

• 用first("Query")是因为每个用户的同一个序号仅对应一条查询（即使有重复搜索内容，也取第一条）；如果需要保留所有重复查询，可以改用collect_list("Query")将同一序号的查询存为数组，但这会改变你期望的单值列结构。
• 如果不同用户的查询数量差异较大，转置后会生成大量含空值的列，这是正常现象，PySpark会自动处理这些空值。

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