Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/6297#discussion_r33001595
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregates.scala
---
@@ -746,3 +746,219 @@ case class LastFunction(expr: Expression, base:
AggregateExpression) extends Agg
if (result != null) expr.eval(result.asInstanceOf[Row]) else null
}
}
+
+// Compute standard deviation based on online algorithm specified here:
+// http://en.wikipedia.org/wiki/Algorithms_for_calculating_variance
+case class Stddev(child: Expression) extends PartialAggregate with
trees.UnaryNode[Expression] {
+
+ override def nullable: Boolean = true
+ override def dataType: DataType = child.dataType match {
+ case DecimalType.Fixed(_, _) | DecimalType.Unlimited =>
+ DecimalType.Unlimited
+ case _=>
+ DoubleType
+ }
+ override def toString: String = s"STDDEV($child)"
+ override def asPartial: SplitEvaluation = {
+ val partialStd = Alias(ComputePartialStd(Cast(child, dataType)),
"PartialStddev")()
+ SplitEvaluation(CombinePartialStd(partialStd.toAttribute), partialStd
:: Nil)
+ }
+ override def newInstance(): StddevFunction = new StddevFunction(child,
this)
+}
+
+case class ComputePartialStd(child: Expression) extends
AggregateExpression {
+ def this() = this(null)
+
+ override def children: Seq[Expression] = child :: Nil
+ override def nullable: Boolean = false
+ override def dataType: DataType = child.dataType match {
+ case DecimalType.Unlimited => ArrayType(DecimalType.Unlimited)
+ case _ => ArrayType(DoubleType)
+ }
+ override def toString: String = s"computePartialStddev($child)"
+ override def newInstance(): ComputePartialStdFunction =
+ new ComputePartialStdFunction(child, this)
+}
+
+case class CombinePartialStd(child: Expression) extends
AggregateExpression {
+ def this() = this(null)
+
+ override def children: Seq[Expression] = child:: Nil
+ override def nullable: Boolean = false
+ override def dataType: DataType = child.dataType match {
+ case ArrayType(DecimalType.Unlimited, _) => DecimalType.Unlimited
+ case _ => DoubleType
+ }
+ override def toString: String = s"CombinePartialStd($child)"
+ override def newInstance(): CombinePartialStdFunction = {
+ new CombinePartialStdFunction(child, this)
+ }
+}
+
+case class ComputePartialStdFunction (
+ expr: Expression,
+ base: AggregateExpression
+) extends AggregateFunction {
+ def this() = this(null, null) // Required for serialization
+
+ private val computeType = expr.dataType
+ private val zero = Cast(Literal(0), computeType)
+ private var partialCount: Long = 0L
+
+ // the mean of data processed so far
+ private val partialAvg :MutableLiteral = MutableLiteral(zero.eval(null),
computeType)
+
+ // update average based on this formula:
+ // avg = avg + (value - avg)/count
+ private def avgAddFunction (value: Literal) : Expression= {
+ val delta = Subtract(Cast(value, computeType), partialAvg)
+ Add (partialAvg, Divide(delta, Cast(Literal(partialCount),
computeType)))
+ }
+
+ // the sum of squares of difference from mean
+ private val partialMk :MutableLiteral = MutableLiteral(zero.eval(null),
computeType)
+
+ // update sum of square of difference from mean based on following
formula:
+ // Mk = Mk + (value - preAvg) * (value - updatedAvg)
+ private def mkAddFunction(value: Literal, prePartialAvg: MutableLiteral)
: Expression = {
+ val delta1 = Subtract(Cast(value, computeType), prePartialAvg)
+ val delta2 = Subtract(Cast(value, computeType), partialAvg)
+ Add(partialMk, Multiply(delta1, delta2))
+ }
+
+ override def update(input: Row): Unit = {
+ val evaluatedExpr = expr.eval(input)
+ if (evaluatedExpr != null) {
+ val exprValue = Literal.create(evaluatedExpr, expr.dataType)
+ val prePartialAvg = partialAvg.copy()
+ partialCount += 1
+ partialAvg.update(avgAddFunction(exprValue), input)
+ partialMk.update(mkAddFunction(exprValue, prePartialAvg), input)
+ }
+ }
+
+ override def eval(input: Row): Any = {
+ Seq(Cast(Literal(partialCount), computeType).eval(null),
+ partialAvg.eval(null),
+ partialMk.eval(null))
+ }
+}
+
+case class CombinePartialStdFunction(
+ expr: Expression,
+ base: AggregateExpression
+) extends AggregateFunction {
+ def this() = this (null, null) // Required for serialization
+
+ private val computeType = expr.dataType match {
+ case ArrayType(DecimalType.Unlimited, _) => DecimalType.Unlimited
+ case _ => DoubleType
+ }
+ private val zero = Cast(Literal(0), computeType)
+ private val combineCount = MutableLiteral(zero.eval(null), computeType)
+ private val combineAvg = MutableLiteral(zero.eval(null), computeType)
+ private val combineMk = MutableLiteral(zero.eval(null), computeType)
+
+ private def avgUpdateFunction(preCount: Expression,
--- End diff --
Wrap all arguments with 4 space indent.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
