“按设备分组，取每条记录之前最接近的一次读数，再把没有前序记录的那几条丢掉”——需求一句话，代码却总在同一层 select 里绕不过去：窗口刚算出来的列，过滤时引用不到。报错通常是列无法解析，改成另一写法后又发现结果少了一大截，或者第一次跑与第二次跑的行数不一致。

窗口函数分层：开窗、打标、再过滤

这不是 API 用得别扭，而是语义本来就分两层。窗口函数产出的是一张新列，它的值来自分区内的其它行；而”排除某些行”是对这张列的条件判断，判断必须在列已经算完之后发生。把这条顺序钉死，剩下的都只是写法选择：帧边界怎么定、并列怎么处理、倾斜怎么付账。

一、求值顺序决定过滤是在窗口之前还是之后

一条带窗口的查询，逻辑上的处理次序大致是：先确定输入行集（FROM、关联条件），再做行级过滤（WHERE），然后分组聚合与分组后过滤，接下来才轮到窗口计算，最后才是投影出的列别名给到排序与外层使用。位置差别带来的结论很直接：

  • 写在窗口之前的过滤，改变的是参与开窗的行集合。你先排掉某台设备，分区的累计值、排名、邻居全都跟着变。
  • 写在窗口之后的过滤，只是从已经算好的结果里挑行。开窗用的数据仍然完整，代价照付。

工业数据里这个差别很贵：想要”某设备相邻 N 条之间的间隔”，如果先把不符合条件的行 filter 掉再开窗，剩下的序列中间就有洞，间隔算出来是跨洞的假值；正确做法是全量开窗算间隔，标记完成之后再用标记去筛。同理，”排除异常值”要看你想排除的是”参与计算的它”还是”结果里的它”——前者放在窗口之前，后者放在窗口之后，语义完全不同，而两种写法都语法正确，只能靠人判断。

二、同一层拿不到别名：标准三步

DataFrame 的每个算子都是一个逻辑节点。一次 select 只生成一个投影节点，节点内的列表达式是并列关系，不存在先后，所以你不能在这一层引用同一层刚起的名字；withColumn 则是在现有节点之上叠一层投影，filter 再叠一层条件节点。这就是”总在下一层做”的原因——不是风格偏好，是解析顺序。

标准三步没有任何花样：

from pyspark.sql import Window
from pyspark.sql import functions as F

w = Window.partitionBy("device_id").orderBy("ts")

marked = (
    df.withColumn("prev_ts", F.lag("ts").over(w))
      .withColumn("gap_sec", F.unix_timestamp("ts") - F.unix_timestamp("prev_ts"))
      .withColumn("keep", F.when(F.col("prev_ts").isNotNull() & (F.col("gap_sec") <= 300), 1).otherwise(0))
)

result = marked.filter(F.col("keep") == 1).drop("keep")

三步各自的职责要固定下来：第一步开窗产列，第二步用这些列打标记，第三步按标记筛行。中间列一律起名带前缀（_rn、_gap、_keep），最后 drop 掉，别让它渗进下游表结构。有人试图把三步压成一步，代价是同一个窗口定义在一条表达式里被重复声明——相同窗口通常会被合并成一次计算，但最终要看执行计划，而且可读性与审查难度立刻上升：出问题时没人能对上”哪个条件是筛行、哪个是打标”。

三、三类动态条件：当前行、相邻行、分区聚合值

需求里的”动态”有三种不同来源，写法与坑各不相同。

第一类基于当前行自身的位置，典型是”只看它之前的行”。用 rowsBetween 定义物理帧，边界是相对偏移：0 代表当前行，负数是它前面第几行，正数是后面第几行，另外还有三个专门表达无穷边界的常量。要”排除自身”，上界写 -1 而不是 0。这里最常见的错误是把 0 当”不含当前行”——它恰恰含。

第二类基于相邻行，用 lead、lag 取偏移行的值。要注意三点：边界行取不到邻居时返回空值，这个空值是信息而不是脏数据，不能一律当异常丢掉；ignoreNulls 这类参数是否可用、偏移量与缺省值怎么给，各版本函数签名不一致，落笔前核对自己那版；排序列必须确定，否则邻居是谁由物理顺序决定，两次运行可以给出不同结果。

第三类基于分区聚合值，比如”和组内均值比”。要么用无界帧把聚合值摊到每一行，要么先聚合成组级小表再关联回去。用帧时要清楚默认帧的陷阱：一旦定义了排序，默认帧是按值范围从分区开头到”当前行所在的同级行结束”，也就是说排序值并列的行会一起进入帧。想让累计只到自己这一行，就得显式改用按物理行的帧。按时间值划定前后区间时用范围帧而不是行数帧，边界单位与排序列类型的换算规则以所用版本文档为准——这类错误的表现是整段窗口偏一个固定量，很难从结果上看出不对。

四、去重、取时间上最靠后的一条与”排除自身”的变体

需求说法 窗口写法 过滤放在哪 主要代价
每组保留时间上最靠后的一条 按分区键分组、按时间倒序排序取序号 序号列打完后下一层筛等于 1 一次按分区键的重分布加分区内排序
最靠后那条也不要，只留它的历史 同上，取序号大于 1 同上 同上，另需在结果里明确”不含锚点行”
有并列时全部保留 用名次函数而不是序号函数 名次等于 1 结果行数可能大于预期，行数断言要按并列上限写
与上一次读数比较，丢弃首次读数 取前一行的值，判空后打标记 下一层按标记筛掉首次读数 边界行天然为空，逻辑要单独说明
前后 N 条之内的邻居数 收窄的行帧（如 -3 到 3） 邻居数作为标记列，下一层筛 帧收窄只减少计算量，不减少重分布
组内占比超阈值就整组剔除 无界帧取组内聚合值，或先聚合成小表再关联 阈值判断在下一层 小表可广播，通常比巨大分区的窗口便宜得多

序号函数与名次函数的区别是这一类问题的分水岭：序号一定给每行一个唯一数字，遇到并列时谁排前面取决于排序表达式能不能完全定序；名次把并列行给同一个数字，于是”每组一条”变成”每组若干条”。工业数据里时间戳重复很常见——同一秒写入多条采样——只按时间排序去重，结果就在并列行里随机挑一个。修法是给排序补一个真正唯一的 tie-breaker，或者干脆换成语义更接近需求的”整组保留”。

五、分区数、倾斜与重算

窗口计算需要按分区键把数据重分布，然后在单个任务里对分区排序并保持分区在内存或溢写文件中。于是两个维度决定成本：分区键的基数（决定并行度上限）与单个分区的行数（决定最慢那个任务）。基数太低时，一半数据挤在两个任务里；基数正常但存在热点键（一台一直在上报的设备、一个空值键吞掉所有缺失记录）时，长尾任务把整个阶段拖住。

  • 分区数偏少导致长尾：先按窗口键做一次计数分布观察，必要时把上游按同一批键重分布，让开窗阶段免掉额外一次交换。
  • 热点键导致单任务溢出：把热点单独拆出来处理，或者改成”先按键聚合再关联”的形态，让每个键只贡献一行。
  • 同一份数据既要用窗口结果又要用原始行：这时才考虑缓存。可复用的中间结果用持久化声明，避免整条血缘被重算两遍；要真正截断血缘则用检查点类接口，它把结果落到存储上，代价是一次完整写入，恢复能力取决于落盘位置——落在集群本地盘的那一类，执行器一旦丢失数据就找不回来。血缘很长且下游只有一两个分支时，重复计算往往比截断划算；分支多、上游又贵时反过来。判断依据是这条链上最贵的那一步在哪，别把落盘当默认动作。
  • 窗口拆成关联的时机：当”分区聚合值”只用于阈值比较、而分区本身巨大时，先聚合出组级小表再广播关联，通常比在巨大分区上开窗口便宜，且并行度不受分区键基数限制。

六、验收：逐层数行数（洹拓视角）

这类代码的正确性靠眼睛看不出来，因为它只在行与行的关系上出错。建议的核对顺序：

  1. 造一个十行以内的小样本，覆盖并列时间戳、缺失键、分区只有一行、跨天边界这四种情况。
  2. 每一步之后把中间结果打出来，重点看行号、邻居列、标记列三者的对应关系是否符合预期。
  3. 对每一层做行数断言：开窗不应改变行数，打标记不应改变行数，只有筛行那一步允许变少，并把预期数量写成常量。
  4. 同一份输入连续跑两次，比对结果集的排序后哈希，用来暴露排序不确定导致的随机取行。
  5. 检查被丢弃的行，人工确认每一类都是该丢的；空值行、边界行被顺手筛掉是最常见的静默数据损失。
  6. 最后看执行计划里窗口阶段有没有额外一次交换，有就说明上游的分布与窗口键没对上。

洹拓科技在数据侧的评审里把这套要求收敛成一句话：窗口逻辑不看代码看层次。分层写清楚，代价与语义都能对上；压成一层，出问题时只能靠猜——而猜出来的结论往往把”筛掉脏数据”写成”筛掉了参与计算的数据”，等到下游指标对不上才被发现。

#PySpark #窗口函数 #数据仓库 #SQL 语义 #性能