Spark DataFrame自定义计算标准差函数运行报错原因咨询

问题根因 #

• 你在map算子中对每个列名c执行的df.select(c).na.fill(0).agg(stddev(c))返回的是DataFrame类型，但外层df.select()方法要求传入的参数是Column对象或字符串列名的集合，输入参数类型不匹配，因此触发重载方法匹配报错。
• 原写法对每个列单独触发子查询，执行效率极低，没有利用Spark的并行计算能力。

修正后的自定义函数 #

import org.apache.spark.sql.functions.stddev

def stdDevAllColumns(df: DataFrame): DataFrame = {
  // 全局将所有列的缺失值填充为0
  val filledDf = df.na.fill(0)
  // 批量对所有列计算标准差，重命名结果列避免歧义
  filledDf.agg(
    df.columns.map(c => stddev(c).alias(s"${c}_stddev")): _*
  )
}

可选优化（过滤非数值列） #

如果你的DataFrame包含字符串等非数值类型的列，计算标准差无意义，可以提前过滤列避免执行报错：

import org.apache.spark.sql.functions.col
import org.apache.spark.sql.types.NumericType
import org.apache.spark.sql.functions.stddev

def stdDevAllColumns(df: DataFrame): DataFrame = {
  // 筛选所有数值类型的列
  val numericCols = df.schema.fields.filter(_.dataType.isInstanceOf[NumericType]).map(_.name)
  val filledDf = df.select(numericCols.map(col): _*).na.fill(0)
  filledDf.agg(
    numericCols.map(c => stddev(c).alias(s"${c}_stddev")): _*
  )
}

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