PySpark使用‘count’列名执行聚合查询的报错与解决方法

问题场景 #

在PySpark中对名为count的列执行聚合查询时触发类型错误，最初误以为count是系统保留字，排查后确认是函数导入冲突导致的问题。

原代码示例 #

from pyspark.sql.types import StructType, StructField, IntegerType, TimestampType
from pyspark.sql.functions import sum, avg, max, from_json, col, window

json_schema = StructType([
    StructField("userId", IntegerType(), True),
    StructField("count", IntegerType(), True),
    StructField("dt", TimestampType(), True),
])

# 数据处理逻辑
df \
 .select(from_json(col("value").cast("string"), json_schema).alias("data")) \
 .select("data.*") \
 .groupBy(window(col("dt"), "30 seconds")) \
 .agg(
     sum("count").alias("sum_salary"),
     avg("count").alias("avg_salary"),
     max("count").alias("max_bonus")
 )

报错信息 #

TypeError: unsupported operand type(s) for +: 'int' and 'str'

报错原因 #

如果直接通过from pyspark.sql.functions import sum导入函数，调用sum("count")时Python会优先使用内置的sum函数，而非PySpark提供的聚合sum函数。

Python内置sum要求参数是可迭代对象（比如列表、元组），传入字符串"count"时，它会尝试遍历字符串字符并累加，和聚合逻辑中的整数类型操作冲突，最终触发类型错误。

解决方法 #

通过别名导入PySpark函数库，明确指定使用PySpark的聚合函数：

import pyspark.sql.functions as f

# 修改后的agg部分
.agg(
    f.sum("count").alias("sum_count"),
    f.avg("count").alias("avg_count"),
    f.max("count").alias("max_count")
)

通过f.前缀调用PySpark函数，彻底避免和Python内置函数的命名冲突。

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