PySpark Pandas自定义聚合报错：使用numpy.nanmean时格式不符

错误原因 #

PySpark Pandas（Spark上的Pandas API）的agg方法和原生Pandas不同，它底层依赖Spark分布式执行逻辑，要求聚合函数必须是Spark支持的内置函数名称（字符串形式），无法直接传入Python函数（比如np.nanmean），这就是你收到"aggs must be a dict mapping from column name to aggregate functions (string or list of strings)"报错的原因。

解决方案 #

方法1：用Spark内置函数替代（优先推荐）

Spark的avg函数会自动忽略null值，和numpy.nanmean的效果完全一致，直接用字符串'avg'即可实现需求：

import pandas as pd
import pyspark.pandas as ps
import numpy as np

# 创建示例PySpark Pandas DataFrame
data = {'Category': ['A', 'B', 'A', 'B', 'A', 'B'],
        'Value1': [10, 20, 30, 40, 50, 60],
        'Value2': [100, np.nan, 300, 400, 500, 600]}  
sdf = spark.createDataFrame(pd.DataFrame(data))
pdf = ps.DataFrame(sdf)

# 使用Spark内置聚合函数
result = pdf.groupby('Category').agg(
    Sum_Value1=('Value1', 'sum'),
    Mean_val2=('Value2', 'avg')
)

print(result)

输出结果：

Sum_Value1  Mean_val2
Category                      
A                90      300.0
B               120      500.0

方法2：自定义聚合函数（适用于非内置函数场景）

如果必须使用自定义Python逻辑（比如复杂的自定义聚合），可以用groupby.apply方法，在每个分组上执行Pandas本地聚合：

import pandas as pd
import pyspark.pandas as ps
import numpy as np

# 创建示例数据
data = {'Category': ['A', 'B', 'A', 'B', 'A', 'B'],
        'Value1': [10, 20, 30, 40, 50, 60],
        'Value2': [100, np.nan, 300, 400, 500, 600]}  
sdf = spark.createDataFrame(pd.DataFrame(data))
pdf = ps.DataFrame(sdf)

# 定义分组聚合逻辑，输入为单个分组的Pandas DataFrame
def custom_group_agg(df):
    return pd.Series({
        'Sum_Value1': df['Value1'].sum(),
        'Mean_val2': np.nanmean(df['Value2'])
    })

# 应用自定义聚合
result = pdf.groupby('Category').apply(custom_group_agg)

print(result)

注意事项 #

• 优先使用Spark内置聚合函数，性能更优，适合大规模分布式数据处理。
• groupby.apply会将分组数据拉到本地节点执行Pandas操作，仅适合分组数据量较小的场景，避免内存溢出。

内容的提问来源于stack exchange，提问作者Diaa Al mohamad