Spark窗口函数使用AVG报错：无法提取Value值的解决方法

问题背景 #

原始SQL查询 #

Select st.Value,
       st.Id,
       ntile(2) OVER (PARTITION BY St.Id, St.VarId ORDER By St.Sls),
       AVG(St.Value) OVER (PARTITION BY St.Id, St.VarId ORDER By St.Sls, St.Date)
FROM table tb
INNER JOIN staging st on St.Id = tb.Id

转换后的PySpark代码 #

windowSpec_1 = Window.partitionBy("staging.Id", "staging.VarId").orderBy("staging.Sls")
windowSpec_2 = Window.partitionBy("staging.Id", "staging.VarId").orderBy("staging.Sls", "staging.Date")

df= table.join(
    staging,
    on=f.col("staging.Id") == f.col("table.Id"),
    how='inner'
).select(
    f.col("staging.Value"),
    f.ntile(2).over(windowSpec_1),
    f.avg("staging.Value").over(windowSpec_2)
)

报错信息 #

pyspark.sql.utils.AnalysisException: Can't extract value from Value#42928: need struct type but got decimal(16,6)

错误原因与解决方案 #

核心原因 #

报错是因为表join后出现重复列名：如果table和staging表都存在名为Value的列，Spark会自动将重复列封装为struct类型（比如Value字段变成包含两个子列的结构体）。此时用f.col("staging.Value")引用时，Spark会错误地认为要从这个struct中提取staging字段，但实际原始的staging.Value是decimal类型，因此触发类型不匹配错误。

解决方法 #

不需要额外分组，只需消除列名冲突，确保引用的列是明确的非struct类型，以下是两种可行方案：

方案1：给表设置别名并明确引用列

from pyspark.sql import functions as f
from pyspark.sql.window import Window

# 为表设置别名，避免列名混淆
tb = table.alias("tb")
st = staging.alias("st")

# 定义窗口规则（使用别名引用列）
windowSpec_1 = Window.partitionBy("st.Id", "st.VarId").orderBy("st.Sls")
windowSpec_2 = Window.partitionBy("st.Id", "st.VarId").orderBy("st.Sls", "st.Date")

df = tb.join(
    st,
    on=f.col("st.Id") == f.col("tb.Id"),
    how='inner'
).select(
    f.col("st.Value"),
    f.col("st.Id"),
    f.ntile(2).over(windowSpec_1).alias("ntile_group"),
    f.avg(f.col("st.Value")).over(windowSpec_2).alias("avg_value")
)

方案2：join前筛选列，避免引入重复列

如果只需要table表中的Id列用于关联，可以提前筛选列，避免重复列名：

from pyspark.sql import functions as f
from pyspark.sql.window import Window

# 只保留需要的列，避免重复列
tb_selected = table.select("Id").alias("tb")
st_selected = staging.select("Id", "VarId", "Sls", "Date", "Value").alias("st")

windowSpec_1 = Window.partitionBy("st.Id", "st.VarId").orderBy("st.Sls")
windowSpec_2 = Window.partitionBy("st.Id", "st.VarId").orderBy("st.Sls", "st.Date")

df = tb_selected.join(
    st_selected,
    on="Id",  # 两边仅保留Id列，可直接用列名关联
    how='inner'
).select(
    f.col("st.Value"),
    f.col("st.Id"),
    f.ntile(2).over(windowSpec_1),
    f.avg(f.col("st.Value")).over(windowSpec_2)
)

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