pySpark 在忽略 NaN 的同时计算多列的最大值

Question

我正在尝试计算 pyspark 数据框中多列的最大值（或任何聚合函数）。但是，由于这些列有一些 NaN，因此 max 聚合器的结果始终为 NaN。有什么方法可以在进行此计算时忽略 NaN 吗？

这是我的代码运行:

panel = pd.DataFrame({'col1': [2,3,4,np.nan],
                         'col2': [1,np.nan,4,np.nan],
                         'col3': [2,7,1,np.nan],
                         'col4': [np.nan,3,4,np.nan]})
sq = context.sql
sparkDF = sq.createDataFrame(panel) 

# Compute max
mapp = {c: 'max' for c in panel.columns} # {'col1': 'max', col2': 'max', ...}
vals = panel.agg(mapp).collect()[0]
print(vals)

PS：我不能只删除包含 NaN 的行，因为其他列可能具有有效值。我还想避免必须遍历每一列并计算最大值 'manually'，因为这会导致很大的性能影响。

Answer 1

您可以尝试 max 和 fill。

from pyspark.sql import functions as F
sparkDF = sparkDF.na.fill(0).select([F.max(x).alias(f'max_{x}') for x in sparkDF.columns])

sparkDF.show()

# +--------+--------+--------+--------+
# |max_col1|max_col2|max_col3|max_col4|
# +--------+--------+--------+--------+
# |     4.0|     4.0|     7.0|     4.0|
# +--------+--------+--------+--------+

=========================================== =======

更新：

要计算忽略 NaN 的平均值，您不能使用常规 avg，因为这包括 NaN。为了通过删除 NaN 来计算平均值，您可以在条件下使用 sum 和 count。

sparkDF = sparkDF.select([
    (F.sum(F.when(~F.isnan(x), F.col(x))) / F.count(F.when(~F.isnan(x), True))).alias(f'avg_{x}')
    for x in sparkDF.columns
])

pySpark 在忽略 NaN 的同时计算多列的最大值

pySpark compute max of multiple columns while ignoring NaN

python

dataframe

pyspark

apache-spark-sql

aggregate