Spark DataFrame：计算按行均值（或任何聚合操作）

  • Z时代
  • 2024-01-10
  • 分类：问答

我在内存中加载了一个Spark DataFrame，我想对各列取均值（或任何聚合操作）。我该怎么办？（在中numpy，这称为接管操作axis=1）。

如果正在计算行（axis=0）下方的DataFrame平均值，则该数据已内置：

from pyspark.sql import functions as F
F.mean(...)

但是，有没有一种方法可以针对列中的条目以编程方式执行此操作？例如，从下面的DataFrame

+--+--+---+---+
|id|US| UK|Can|
+--+--+---+---+
| 1|50|  0|  0|
| 1| 0|100|  0|
| 1| 0|  0|125|
| 2|75|  0|  0|
+--+--+---+---+

省略id，意味着

+------+
|  mean|
+------+
| 16.66|
| 33.33|
| 41.67|
| 25.00|
+------+

回答：

您所需要做的就是这样的标准SQL：

SELECT (US + UK + CAN) / 3 AS mean FROM df

可以直接与SqlContext.sqlDSL一起使用或通过DSL表达

df.select(((col("UK") + col("US") + col("CAN")) / lit(3)).alias("mean"))

如果您有更多的列，则可以如下生成表达式：

from functools import reduce
from operator import add
from pyspark.sql.functions import col, lit
n = lit(len(df.columns) - 1.0)
rowMean  = (reduce(add, (col(x) for x in df.columns[1:])) / n).alias("mean")
df.select(rowMean)

要么

rowMean  = (sum(col(x) for x in df.columns[1:]) / n).alias("mean")
df.select(rowMean)

最后，它在Scala中等效：

df.select(df.columns
  .drop(1)
  .map(col)
  .reduce(_ + _)
  .divide(df.columns.size - 1)
  .alias("mean"))

在更复杂的情况下，您可以使用array函数组合列并使用UDF计算统计信息：

import numpy as np
from pyspark.sql.functions import array, udf
from pyspark.sql.types import FloatType
combined = array(*(col(x) for x in df.columns[1:]))
median_udf = udf(lambda xs: float(np.median(xs)), FloatType())
df.select(median_udf(combined).alias("median"))

使用Scala API表示的相同操作：

val combined = array(df.columns.drop(1).map(col).map(_.cast(DoubleType)): _*)
val median_udf = udf((xs: Seq[Double]) => 
    breeze.stats.DescriptiveStats.percentile(xs, 0.5))
df.select(median_udf(combined).alias("median"))

以上是 Spark DataFrame：计算按行均值（或任何聚合操作） 的全部内容， 来源链接： utcz.com/qa/406673.html

回到顶部