在这种情况下，您不会使用lead()和lag()，而是使用聚合。在这种情况下，您还需要一些条件聚合。
这可能是最好的方法：

select t.entity_id, t.num_signals,
       tmin.item_id as item_id_min_month,
       tmax.item_id as item_id_max_month
from (select t.entity_id, sum(signal_count) as num_signals,
             min(month_id) as min_month_id,
             max(month_id) as max_month_id
      from t
     ) t join
     t tmin
     on tmin.entity_id = t.entity_id and
        tmin.month_id = t.min_month_id join
     t tmax
     on tmax.entity_id = t.entity_id and
        tmax.month_id = t.max_month_id ;

如果您的SparkSQL版本中提供了first_value()和last_value()，您也可以使用它们。

您需要使用滞后和超前执行以下代码更改

val windowSpecLag = Window.partitionBy("entity_id").orderBy($"month_id".desc_nulls_first, $"item_id".asc_nulls_first)
val windowSpecLead = Window.partitionBy("entity_id").orderBy($"month_id".desc_nulls_first, $"item_id".desc_nulls_first)
val windowSpecSum = Window.partitionBy("entity_id")

val outputResultDf = fileContentAsDf
.withColumn("total_signals", sum(col("signal_count")).over(windowSpecSum))
.withColumn("lag", lag($"item_id", 1).over(windowSpecLag))
.withColumn("lead", lead($"item_id", 1).over(windowSpecLead))
.filter($"lag".isNull || $"lead".isNull)
.select("entity_id", "lag", "lead", "total_signals", "item_id")
.groupBy("entity_id")
.agg(last("item_id", ignoreNulls = true).alias("oldest_item_id"),
  first("item_id", ignoreNulls = true).alias("newest_item_id"),
  first("total_signals").alias("total_signals"))