在Pandas预定义窗口内循环生成条件列结果异常求助

你的循环逻辑是全局维护last_temperature变量，但实际需求是在每个(a,b)分区内保留最近的overshoot=0时的温度值。全局变量会跨分区传递值，导致不同分区的结果互相污染，这是你得到错误结果的核心原因。

下面提供两种可行方案，分别适配Pandas和Spark场景：

方案1：在Pandas中按分区处理 #

利用groupby按a和b分组，在每个分组内执行逻辑，避免跨分区干扰：

def fill_temp_filter(group):
    last_temp = None
    temp_filter = []
    for _, row in group.iterrows():
        if row["overshoot"] == 0:
            last_temp = row["temperature"]
        temp_filter.append(last_temp)
    group["temp_filter"] = temp_filter
    return group

# 按(a,b)分组后应用函数
data_new_pd = data_new_pd.groupby(['a', 'b'], group_keys=False).apply(fill_temp_filter)

更简洁的写法可以结合mask和ffill（向前填充）：

# 先把overshoot≠0的temperature设为NaN，再按组向前填充
data_new_pd["temp_filter"] = (
    data_new_pd
    .groupby(['a', 'b'])['temperature']
    .transform(lambda x: x.mask(data_new_pd['overshoot'] != 0).ffill())
)

方案2：回到Spark中处理（更适合大数据场景） #

既然你最初是用Spark生成的数据集，直接在Spark层完成逻辑可以避免转Pandas的开销，同时原生支持分区窗口操作：

from pyspark.sql import Window
from pyspark.sql.functions import when, last

# 定义分区窗口：按a、b分区，按timestamp排序
window_spec = Window.partitionBy('a', 'b').orderBy('timestamp').rowsBetween(Window.unboundedPreceding, Window.currentRow)

data_new_spark = (
    data_pd
    .filter(col('x') > 1)
    .filter(col('x') < 21)
    .withColumn('row', row_number().over(Window.partitionBy('a', 'b').orderBy('timestamp')))
    .filter(col('row') == 1)
    # 新增temp_filter列：仅保留overshoot=0的temperature，否则用之前最近的有效值
    .withColumn('temp_filter', 
                last(when(col('overshoot') == 0, col('temperature')), ignorenulls=True).over(window_spec)
               )
)

# 如需转Pandas再执行
data_new_pd = data_new_spark.toPandas()

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