疑似Bug：UDAF与内置聚合函数共存引发NullPointerException

结论 #

这并非你的用法错误，而是PyFlink在处理Pandas类型UDAF与内置聚合函数混合窗口聚合时存在的兼容性bug。

问题原因 #

当在同一个GROUP BY窗口聚合的SELECT语句中，同时使用Pandas UDAF和内置聚合函数（如MAX）时，PyFlink执行引擎在数据传递与算子融合过程中容易触发空指针异常，尤其是两者针对同一字段聚合时更易出现。

解决方案 #

方案1：拆分聚合逻辑，分两步计算 #

先单独计算内置聚合结果，再与UDAF聚合结果关联：

# 第一步：计算内置MAX聚合
agg_max = t_env.from_path('src_trade')\
    .window(Tumble.over(lit(1).seconds).on(col('time')).alias('w'))\
    .group_by(col('id'), col('w'))\
    .select(col('id'), col('w').end.alias('window_end'), col('price').max.alias('max_price'))

# 第二步：计算UDAF聚合
agg_udaf = t_env.from_path('src_trade')\
    .window(Tumble.over(lit(1).seconds).on(col('time')).alias('w'))\
    .group_by(col('id'), col('w'))\
    .select(col('id'), col('w').end.alias('window_end'), 
            last_value_agg(col('price')).alias('last_price'),
            first_value_agg(col('price')).alias('first_price'))

# 关联两个结果
result = agg_udaf.join(agg_max, 
                       (agg_udaf.id == agg_max.id) & (agg_udaf.window_end == agg_max.window_end))\
    .select(agg_udaf.id, agg_udaf.window_end, agg_udaf.last_price, agg_udaf.first_price, agg_max.max_price)

result.execute().print()

方案2：替换Pandas UDAF为内置函数或普通UDAF #

• 你的first_value_agg存在逻辑错误，当前返回序列最后一个元素（v.iloc[-1]），应改为v.iloc[0]
• 直接使用PyFlink内置函数替代自定义UDAF：

from pyflink.table.window import Window
from pyflink.table.functions import last_value, first_value

(
    t_env.from_path('src_trade')
    .window(Tumble.over(lit(1).seconds).on(col('time')).alias('w'))
    .group_by(col('id'), col('w'))
    .select(
        col('id'),
        col('w').end,
        last_value(col('price')).over(Window.partition_by(col('id')).order_by(col('time')).rows_between(-999999, 0)).alias('last_price'),
        first_value(col('price')).over(Window.partition_by(col('id')).order_by(col('time')).rows_between(-999999, 0)).alias('first_price'),
        col('price').max
    )
    .execute().print()
)

方案3：升级PyFlink版本 #

该问题在PyFlink 1.17及以上的新版本中大概率已被修复，升级到最新稳定版可直接解决此问题。

额外修正提示 #

你的first_value_agg函数实现与命名不符，需修正逻辑：

@udaf(result_type=DataTypes.INT(), func_type='pandas')
def first_value_agg(v: pd.Series):
    return v.iloc[0]  # 原代码错误返回了最后一个元素

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