听起来像是将“第一”行定义为组中最小值为col1的行，将“最后”行定义为组中最大值为col1的行。要计算它们，可以使用MIN和MAX窗口函数：

SELECT
    ID,
    col1,
    (MAX(col1) OVER (PARTITION BY ID)) - (MIN(col1) OVER (PARTITION BY ID)) AS col2
FROM
    ...

如果定义“第一行”和“最后一行”的方式不同（例如，根据某个时间戳），则可以使用更通用的FIRST_VALUE和LAST_VALUE窗口函数：

SELECT
    ID,
    col1,
    (LAST_VALUE(col1) OVER (PARTITION BY ID ORDER BY col1 ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING))
    -
    (FIRST_VALUE(col1) OVER (PARTITION BY ID ORDER BY col1 ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING))
    AS col2
FROM
    ...

上面的两个片段是等效的，但后者更通用：您可以指定按不同的列排序和/或修改窗口规格。

代码如下

w=Window.partitionBy('ID').orderBy('col1').rowsBetween(Window.unboundedPreceding, Window.unboundedFollowing)
    df.withColumn('out', last('col1').over(w)-first('col1').over(w)).show()