spark sql 不支持在 over 子句中直接调用 udf，因执行计划层面仅接受内置窗口函数；必须拆分为两步：先用 udf 生成新列，再在其上应用窗口函数。
不能直接在 OVER 子句里调用 UDF —— 这是 Spark SQL 的硬性限制，不是写法问题，而是执行计划层面不支持。
UDF 无法嵌套进窗口函数的 OVER 子句
你可能会尝试写类似这样的 SQL：
SELECT name, score, my_udf(score) OVER (PARTITION BY dept ORDER BY score DESC) FROM emp
这会报错：UnsupportedOperationException: Expression 'my_udf(score)' not supported within window function。Spark 在解析 OVER 时只接受内置聚合/分析函数（如 row_number()、sum()、lag()），不接受任何 UDF。
- UDF 是运行时动态调用的 JVM/Python 函数，而窗口函数的执行依赖 Catalyst 的静态分析和物理计划优化
- 即使 UDF 是确定性的（
deterministic = True），也无法绕过该限制 - PySpark 中用
pyspark.sql.functions.udf注册的函数同样受此约束
可行路径：先计算再开窗，或先开窗再计算
必须把 UDF 和窗口逻辑拆成两步，用中间列衔接。核心原则是：窗口函数只能作用于「已物化」的列（包括 UDF 输出列）。
- 若需对原始字段做转换后再排序/分组：先用 UDF 生成新列，再在该列上用
row_number()等 - 若需对窗口结果进一步加工（如把排名转为等级描述）：先算
rank()，再用 UDF 映射为 "Top3" / "Others" - DSL 风格更灵活：可用
withColumn("score_norm", my_udf(col("score"))).withColumn("rn", row_number().over(w))
例如，将分数归一化后取部门内 Top3：
from pyspark.sql import functions as F
from pyspark.sql.window import Window
norm_udf = F.udf(lambda x: x / 100.0 if x else 0.0)
w = Window.partitionBy("dept").orderBy(F.col("score_norm").desc())
df.withColumn("score_norm", norm_udf(F.col("score"))) \
.withColumn("rn", F.row_number().over(w)) \
.filter(F.col("rn")
<h3>注意 UDF 类型与窗口函数输出类型的兼容性</h3>
<p>UDF 返回类型必须能被窗口函数后续操作接受。常见踩坑点：</p>
- 返回
None或空字符串 → 窗口函数可能跳过整行（取决于 null 处理策略） - UDF 返回 list/dict →
row_number()无法在其上排序，会报类型错误 - 用 Pandas UDF（
pandas_udf(returnType=...)）时，确保返回 Series 且索引对齐，否则窗口计算结果错位 - 时间类 UDF（如解析字符串为
timestamp）后，再用lead()计算时间差，必须确认返回的是TimestampType，而非字符串
性能敏感场景：优先用内置函数替代 UDF + 窗口组合
每多一层 UDF 就多一次 JVM/Python 进程间序列化开销，叠加窗口计算极易成为瓶颈。
- 字符串截取、大小写转换、数值四则运算等，一律用
F.substring()、F.upper()、F.col("a") + F.col("b")替代 UDF - 条件映射尽量用
F.when()+F.otherwise()，比 Python UDF 快 3–5 倍 - 涉及复杂逻辑（如正则提取多组命名捕获）且必须用 UDF 时，改用
pandas_udf并开启 Arrow 优化（spark.sql.execution.arrow.pyspark.enabled=true）
真正难处理的，是那些既需要自定义逻辑、又强依赖窗口上下文的场景——比如“每个用户最近 3 次订单中，金额最高的那个订单的配送地址是否含‘保税区’”。这种必须拆成两层：先窗口取 top3，再 UDF 判断地址，中间不能省略物化步骤。
