PySpark SQL select语句中如何基于动态列列表生成列并结合窗口函数取首行id

核心逻辑非常简单，全程不需要手动逐列写select表达式，只需要维护一个列名列表，窗口分区规则、select返回列都会自动同步匹配。

实现逻辑要点 #

• Window.partitionBy原生支持接收列名字符串列表作为入参，不需要逐个传col对象，直接把动态列表丢进去就能完成分区规则的调整。
• PySpark的select方法支持直接接收字符串格式的列名，和手动写col("列名")的效果完全一致。借助Python的列表拆解语法，直接把动态列列表拆成参数传入select，就能自动生成所有需要的分区列逻辑。
• 后续如果要调整分区列（比如新增weight、manufacturer等列进规则），只需要修改维护的列名列表，不需要改动select、窗口的任何代码，完全满足需求。

完整可运行代码 #

已修正原有代码里的括号错位语法问题：

from pyspark.sql import Window
from pyspark.sql.functions import first

# 唯一需要动态维护的配置：调整这个列表即可同步所有逻辑
column_list = [
    'productname', 
    'productbrand', 
    'producttype',
    'weight'
]

# 窗口定义：直接传入动态列列表做分区
w2 = (
    Window()
    .partitionBy(column_list)
    .orderBy(unique_attributes.id.asc())
)

# 动态select：拆解列列表 + 拼接窗口计算的首值列
first_item_id_df = (
    unique_attributes
    .select(
        # 拆解列列表，自动生成所有分区列的选择逻辑
        *column_list,
        # 窗口计算每个分区下的首个id值
        first("id", ignorenulls=True).over(w2).alias("matchid")
    )
    .distinct()
)

后续调整示例 #

如果要新增列到分区规则中，只需要修改column_list的内容即可，其他代码完全不需要改动：

# 仅更新列列表，窗口、select逻辑自动适配
column_list = [
    'productname', 
    'productbrand', 
    'producttype',
    'weight',
    'manufacturer'
]

补充说明：*column_list是Python的列表拆解语法，会把列表中的每个列名单独作为参数传给select方法，和手动逐行写col("productname")、col("productbrand")的执行效果完全等价，不需要额外写循环动态生成col对象。

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