Github user JihongMA commented on a diff in the pull request:
https://github.com/apache/spark/pull/6297#discussion_r34717145
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregates.scala
---
@@ -761,3 +761,216 @@ case class LastFunction(expr: Expression, base:
AggregateExpression) extends Agg
if (result != null) expr.eval(result.asInstanceOf[InternalRow]) 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
--- End diff --
the result of ComputePartialStd is array of (partial count, partial avg,
partial Mk), the input data type of expr will be either Decimal.Unlimited or
Double depending on if Child is Decimal type or any other numeric type.
---
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 [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
