You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by clockfly <gi...@git.apache.org> on 2016/08/29 21:25:19 UTC
[GitHub] spark pull request #14868: Implements percentile_approx aggregation function...
GitHub user clockfly opened a pull request:
https://github.com/apache/spark/pull/14868
Implements percentile_approx aggregation function which supports partial aggregation.
## What changes were proposed in this pull request?
This PR implements aggregation function `percentile_approx`. Function `percentile_approx` returns the approximate percentile(s) of a column at the given percentage(s). A percentile is a watermark value below which a given percentage of the column values fall. For example, the percentile of column `col` at percentage 50% is the median value of column `col`.
### Syntax:
```
# Returns percentile at a given percentage value. The approximation error can be reduced by increasing parameter accuracy, at the cost of memory.
percentile_approx(col, percentage [, accuracy])
# Returns percentile value array at given percentage value array
percentile_approx(col, array(percentage1 [, percentage2]...) [, accuracy])
```
### Features:
1. This function supports partial aggregation.
2. The memory consumption is bounded. The larger `accuracy` parameter we choose, we smaller error we get. The default accuracy value is 10000, to match with Hive default setting. Choose a smaller value for smaller memory footprint.
3. This function supports window function aggregation.
### Example usages:
```
## Returns the 25th percentile value, with default accuracy
SELECT percentile_approx(col, 0.25) FROM table
## Returns an array of percentile value (25th, 50th, 75th), with default accuracy
SELECT percentile_approx(col, array(0.25, 0.5, 0.75)) FROM table
## Returns 25th percentile value, with custom accuracy value 100, larger accuracy parameter yields smaller approximation error
SELECT percentile_approx(col, 0.25, 100) FROM table
## Returns the 25th, and 50th percentile values, with custom accuracy value 100
SELECT percentile_approx(col, array(0.25, 0.5), 100) FROM table
```
### NOTE:
1. The `percentile_approx` implementation is different from Hive, so the result returned on same query maybe slightly different with Hive. This implementation uses `QuantileSummaries` as the underlying probabilistic data structure, and mainly follows paper `Space-efficient Online Computation of Quantile Summaries` by Greenwald, Michael and Khanna, Sanjeev. (http://dx.doi.org/10.1145/375663.375670)`
2. The current implementation of `QuantileSummaries` doesn't support automatic compression. This PR has a rule to do compression automatically at the caller side, but it may not be optimal.
## How was this patch tested?
Unit test, and Sql query test.
## Acknowledgement
1. This PR's work in based on @lw-lin's PR https://github.com/apache/spark/pull/14298, with improvements like supporting partial aggregation, fixing out of memory issue.
2. Thanks to @cloud-fan for the implementation of serialization using Expression Encoder.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/clockfly/spark appro_percentile_try_2
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/14868.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #14868
----
commit d71397cc283493a946c42124e41309d7e64d0b12
Author: Sean Zhong <se...@databricks.com>
Date: 2016-08-19T16:34:56Z
percentile approximate
----
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: Implements percentile_approx aggregation function which ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14868
Merged build finished. Test FAILed.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14868: Implements percentile_approx aggregation function...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/14868#discussion_r76782663
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/ApproximatePercentile.scala ---
@@ -0,0 +1,304 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions.aggregate
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.{InternalRow, ScalaReflectionLock}
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.{TypeCheckFailure, TypeCheckSuccess}
+import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.ApproximatePercentile.{PercentileDigest, PercentileDigestSerializer}
+import org.apache.spark.sql.catalyst.util.{ArrayData, GenericArrayData}
+import org.apache.spark.sql.catalyst.util.QuantileSummaries
+import org.apache.spark.sql.catalyst.util.QuantileSummaries.{defaultCompressThreshold, Stats}
+import org.apache.spark.sql.types._
+
+/**
+ * The ApproximatePercentile function returns the approximate percentile(s) of a column at the given
+ * percentage(s). A percentile is a watermark value below which a given percentage of the column
+ * values fall. For example, the percentile of column `col` at percentage 50% is the median of
+ * column `col`.
+ *
+ * This function supports partial aggregation.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param percentageExpression Expression that represents a single percentage value or
+ * a array of percentage values. Each percentage value must be between
+ * 0.0 and 1.0.
+ * @param accuracyExpression Integer literal expression of approximation accuracy. Higher value
+ * yields better accuracy, the default value is
+ * DEFAULT_PERCENTILE_ACCURACY.
+ */
+@ExpressionDescription(
+ usage =
+ """
+ _FUNC_(col, percentage [, accuracy]) - Returns the approximate percentile value of numeric
+ column `col` at the given percentage. The value of percentage must be between 0.0
+ and 1.0. The `accuracy` parameter (default: 10000) is a positive integer literal which
+ controls approximation accuracy at the cost of memory. Higher value of `accuracy` yields
+ better accuracy, `1.0/accuracy` is the relative error of the approximation.
+
+ _FUNC_(col, array(percentage1 [, percentage2]...) [, accuracy]) - Returns the approximate
+ percentile array of column `col` at the given percentage array. Each value of the
+ percentage array must be between 0.0 and 1.0. The `accuracy` parameter (default: 10000) is
+ a positive integer literal which controls approximation accuracy at the cost of memory.
+ Higher value of `accuracy` yields better accuracy, `1.0/accuracy` is the relative error of
+ the approximation.
+ """)
+case class ApproximatePercentile(
+ child: Expression,
+ percentageExpression: Expression,
+ accuracyExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int) extends TypedImperativeAggregate[PercentileDigest] {
+
+ def this(child: Expression, percentageExpression: Expression, accuracyExpression: Expression) = {
+ this(child, percentageExpression, accuracyExpression, 0, 0)
+ }
+
+ def this(child: Expression, percentageExpression: Expression) = {
+ this(child, percentageExpression, Literal(ApproximatePercentile.DEFAULT_PERCENTILE_ACCURACY))
+ }
+
+ // Mark as lazy so that accuracyExpression is not evaluated during tree transformation.
+ private lazy val accuracy: Int = accuracyExpression.eval().asInstanceOf[Int]
+
+ private lazy val serializer: PercentileDigestSerializer = new PercentileDigestSerializer
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(DoubleType, TypeCollection(DoubleType, ArrayType), IntegerType)
+ }
+
+ // Mark as lazy so that percentageExpression is not evaluated during tree transformation.
+ private lazy val (returnPercentileArray: Boolean, percentages: Array[Double]) = {
+ (percentageExpression.dataType, percentageExpression.eval()) match {
+ // Rule ImplicitTypeCasts can cast other numeric types to double
+ case (_, num: Double) => (false, Array(num))
+ case (ArrayType(baseType: NumericType, _), arrayData: ArrayData) =>
+ val numericArray = arrayData.toObjectArray(baseType)
+ (true, numericArray.map { x =>
+ baseType.numeric.toDouble(x.asInstanceOf[baseType.InternalType])
+ })
+ case other =>
+ throw new AnalysisException(s"Invalid data type ${other._1} for parameter percentage")
+ }
+ }
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (accuracy <= 0) {
+ TypeCheckFailure(
+ s"The accuracy provided must be a positive integer literal (current value = $accuracy)")
+ } else if (percentages.exists(percentage => percentage < 0.0D || percentage > 1.0D)) {
+ TypeCheckFailure(
+ s"All percentage values must be between 0.0 and 1.0 " +
+ s"(current = ${percentages.mkString(", ")})")
+ } else {
+ TypeCheckSuccess
+ }
+ }
+
+ override def createAggregationBuffer(): PercentileDigest = {
+ val relativeError = 1.0D / accuracy
+ new PercentileDigest(relativeError)
+ }
+
+ override def update(buffer: PercentileDigest, inputRow: InternalRow): Unit = {
+ val value = child.eval(inputRow)
+ // Ignore empty rows, for example: percentile_approx(null)
+ if (value != null) {
+ buffer.add(value.asInstanceOf[Double])
+ }
+ }
+
+ override def merge(buffer: PercentileDigest, other: PercentileDigest): Unit = {
+ buffer.merge(other)
+ }
+
+ override def eval(buffer: PercentileDigest): Any = {
+ val result = buffer.getPercentiles(percentages)
+ if (result.length == 0) {
+ null
+ } else if (returnPercentileArray) {
+ new GenericArrayData(result)
+ } else {
+ result(0)
+ }
+ }
+
+ override def withNewMutableAggBufferOffset(newOffset: Int): ApproximatePercentile =
+ copy(mutableAggBufferOffset = newOffset)
+
+ override def withNewInputAggBufferOffset(newOffset: Int): ApproximatePercentile =
+ copy(inputAggBufferOffset = newOffset)
+
+ override def children: Seq[Expression] = Seq(child, percentageExpression, accuracyExpression)
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = {
+ if (returnPercentileArray) ArrayType(DoubleType) else DoubleType
+ }
+
+ override def prettyName: String = "percentile_approx"
+
+ override def serialize(obj: PercentileDigest): Array[Byte] = {
+ serializer.serialize(obj)
+ }
+
+ override def deserialize(bytes: Array[Byte]): PercentileDigest = {
+ serializer.deserialize(bytes)
+ }
+}
+
+object ApproximatePercentile {
+
+ // Default accuracy of Percentile approximation. Larger value means better accuracy.
+ // The default relative error can be deduced by defaultError = 1.0 / DEFAULT_PERCENTILE_ACCURACY
+ val DEFAULT_PERCENTILE_ACCURACY: Int = 10000
+
+ /**
+ * PercentileDigest is a probabilistic data structure used for approximating percentiles
+ * with limited memory. PercentileDigest is backed by [[QuantileSummaries]].
+ *
+ * @param summaries underlying probabilistic data structure [[QuantileSummaries]].
+ * @param isCompressed An internal flag from class [[QuantileSummaries]] to indicate whether the
+ * underlying quantileSummaries is compressed.
+ */
+ class PercentileDigest(
+ private var summaries: QuantileSummaries,
+ private var isCompressed: Boolean) {
+
+ // Trigger compression if the QuantileSummaries's buffer length exceeds
+ // compressThresHoldBufferLength. The buffer length can be get by
+ // quantileSummaries.sampled.length
+ private[this] final val compressThresHoldBufferLength: Int = {
+ // Max buffer length after compression.
+ val maxBufferLengthAfterCompression: Int = (1 / summaries.relativeError).toInt * 2
+ // A safe upper bound for buffer length before compression
+ maxBufferLengthAfterCompression * 2
+ }
+
+ def this(relativeError: Double) = {
+ this(new QuantileSummaries(defaultCompressThreshold, relativeError), isCompressed = true)
+ }
+
+ /** Returns compressed object of [[QuantileSummaries]] */
+ def quantileSummaries: QuantileSummaries = {
+ if (!isCompressed) compress()
+ summaries
+ }
+
+ /** Insert an observation value into the PercentileDigest data structure. */
+ def add(value: Double): Unit = {
+ summaries = summaries.insert(value)
+ // The result of QuantileSummaries.insert is un-compressed
+ isCompressed = false
+
+ // Currently, QuantileSummaries ignores the construction parameter compressThresHold,
+ // which may cause QuantileSummaries to occupy unbounded memory. We have to hack around here
+ // to make sure QuantileSummaries doesn't occupy infinite memory.
+ // TODO: Figure out why QuantileSummaries ignores construction parameter compressThresHold
+ if (summaries.sampled.length >= compressThresHoldBufferLength) compress()
+ }
+
+ /** In-place merges in another PercentileDigest. */
+ def merge(other: PercentileDigest): Unit = {
+ if (!isCompressed) compress()
+ summaries = summaries.merge(other.quantileSummaries)
+ }
+
+ /**
+ * Returns the approximate percentiles of all observation values at the given percentages.
+ * A percentile is a watermark value below which a given percentage of observation values fall.
+ * For example, the following code returns the 25th, median, and 75th percentiles of
+ * all observation values:
+ *
+ * {{{
+ * val Array(p25, median, p75) = percentileDigest.getPercentiles(Array(0.25, 0.5, 0.75))
+ * }}}
+ */
+ def getPercentiles(percentages: Array[Double]): Array[Double] = {
+ if (!isCompressed) compress()
--- End diff --
is it possible to evaluate a uncompressed QuantileSummaries? Or it will be bad if we use percentile_approx as window function.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: Implements percentile_approx aggregation function which ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14868
Merged build finished. Test FAILed.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14868: Implements percentile_approx aggregation function...
Posted by liancheng <gi...@git.apache.org>.
Github user liancheng commented on a diff in the pull request:
https://github.com/apache/spark/pull/14868#discussion_r76761445
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/ApproximatePercentile.scala ---
@@ -0,0 +1,304 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions.aggregate
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.{InternalRow, ScalaReflectionLock}
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.{TypeCheckFailure, TypeCheckSuccess}
+import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.ApproximatePercentile.{PercentileDigest, PercentileDigestSerializer}
+import org.apache.spark.sql.catalyst.util.{ArrayData, GenericArrayData}
+import org.apache.spark.sql.catalyst.util.QuantileSummaries
+import org.apache.spark.sql.catalyst.util.QuantileSummaries.{defaultCompressThreshold, Stats}
+import org.apache.spark.sql.types._
+
+/**
+ * The ApproximatePercentile function returns the approximate percentile(s) of a column at the given
+ * percentage(s). A percentile is a watermark value below which a given percentage of the column
+ * values fall. For example, the percentile of column `col` at percentage 50% is the median of
+ * column `col`.
+ *
+ * This function supports partial aggregation.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param percentageExpression Expression that represents a single percentage value or
+ * a array of percentage values. Each percentage value must be between
+ * 0.0 and 1.0.
+ * @param accuracyExpression Integer literal expression of approximation accuracy. Higher value
+ * yield better accuracy, the default value is
--- End diff --
Nit: yield => yields
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14868: Implements percentile_approx aggregation function...
Posted by clockfly <gi...@git.apache.org>.
Github user clockfly commented on a diff in the pull request:
https://github.com/apache/spark/pull/14868#discussion_r76898997
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/ApproximatePercentile.scala ---
@@ -0,0 +1,304 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions.aggregate
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.{InternalRow, ScalaReflectionLock}
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.{TypeCheckFailure, TypeCheckSuccess}
+import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.ApproximatePercentile.{PercentileDigest, PercentileDigestSerializer}
+import org.apache.spark.sql.catalyst.util.{ArrayData, GenericArrayData}
+import org.apache.spark.sql.catalyst.util.QuantileSummaries
+import org.apache.spark.sql.catalyst.util.QuantileSummaries.{defaultCompressThreshold, Stats}
+import org.apache.spark.sql.types._
+
+/**
+ * The ApproximatePercentile function returns the approximate percentile(s) of a column at the given
+ * percentage(s). A percentile is a watermark value below which a given percentage of the column
+ * values fall. For example, the percentile of column `col` at percentage 50% is the median of
+ * column `col`.
+ *
+ * This function supports partial aggregation.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param percentageExpression Expression that represents a single percentage value or
+ * a array of percentage values. Each percentage value must be between
+ * 0.0 and 1.0.
+ * @param accuracyExpression Integer literal expression of approximation accuracy. Higher value
+ * yields better accuracy, the default value is
+ * DEFAULT_PERCENTILE_ACCURACY.
+ */
+@ExpressionDescription(
+ usage =
+ """
+ _FUNC_(col, percentage [, accuracy]) - Returns the approximate percentile value of numeric
+ column `col` at the given percentage. The value of percentage must be between 0.0
+ and 1.0. The `accuracy` parameter (default: 10000) is a positive integer literal which
+ controls approximation accuracy at the cost of memory. Higher value of `accuracy` yields
+ better accuracy, `1.0/accuracy` is the relative error of the approximation.
+
+ _FUNC_(col, array(percentage1 [, percentage2]...) [, accuracy]) - Returns the approximate
+ percentile array of column `col` at the given percentage array. Each value of the
+ percentage array must be between 0.0 and 1.0. The `accuracy` parameter (default: 10000) is
+ a positive integer literal which controls approximation accuracy at the cost of memory.
+ Higher value of `accuracy` yields better accuracy, `1.0/accuracy` is the relative error of
+ the approximation.
+ """)
+case class ApproximatePercentile(
+ child: Expression,
+ percentageExpression: Expression,
--- End diff --
And we can use the InputType Check facility to get a more consistent error message if check fails.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14868: Implements percentile_approx aggregation function...
Posted by clockfly <gi...@git.apache.org>.
Github user clockfly commented on a diff in the pull request:
https://github.com/apache/spark/pull/14868#discussion_r76898056
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/ApproximatePercentile.scala ---
@@ -0,0 +1,304 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions.aggregate
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.{InternalRow, ScalaReflectionLock}
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.{TypeCheckFailure, TypeCheckSuccess}
+import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.ApproximatePercentile.{PercentileDigest, PercentileDigestSerializer}
+import org.apache.spark.sql.catalyst.util.{ArrayData, GenericArrayData}
+import org.apache.spark.sql.catalyst.util.QuantileSummaries
+import org.apache.spark.sql.catalyst.util.QuantileSummaries.{defaultCompressThreshold, Stats}
+import org.apache.spark.sql.types._
+
+/**
+ * The ApproximatePercentile function returns the approximate percentile(s) of a column at the given
+ * percentage(s). A percentile is a watermark value below which a given percentage of the column
+ * values fall. For example, the percentile of column `col` at percentage 50% is the median of
+ * column `col`.
+ *
+ * This function supports partial aggregation.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param percentageExpression Expression that represents a single percentage value or
+ * a array of percentage values. Each percentage value must be between
+ * 0.0 and 1.0.
+ * @param accuracyExpression Integer literal expression of approximation accuracy. Higher value
+ * yields better accuracy, the default value is
+ * DEFAULT_PERCENTILE_ACCURACY.
+ */
+@ExpressionDescription(
+ usage =
+ """
+ _FUNC_(col, percentage [, accuracy]) - Returns the approximate percentile value of numeric
+ column `col` at the given percentage. The value of percentage must be between 0.0
+ and 1.0. The `accuracy` parameter (default: 10000) is a positive integer literal which
+ controls approximation accuracy at the cost of memory. Higher value of `accuracy` yields
+ better accuracy, `1.0/accuracy` is the relative error of the approximation.
+
+ _FUNC_(col, array(percentage1 [, percentage2]...) [, accuracy]) - Returns the approximate
+ percentile array of column `col` at the given percentage array. Each value of the
+ percentage array must be between 0.0 and 1.0. The `accuracy` parameter (default: 10000) is
+ a positive integer literal which controls approximation accuracy at the cost of memory.
+ Higher value of `accuracy` yields better accuracy, `1.0/accuracy` is the relative error of
+ the approximation.
+ """)
+case class ApproximatePercentile(
+ child: Expression,
+ percentageExpression: Expression,
+ accuracyExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int) extends TypedImperativeAggregate[PercentileDigest] {
+
+ def this(child: Expression, percentageExpression: Expression, accuracyExpression: Expression) = {
+ this(child, percentageExpression, accuracyExpression, 0, 0)
+ }
+
+ def this(child: Expression, percentageExpression: Expression) = {
+ this(child, percentageExpression, Literal(ApproximatePercentile.DEFAULT_PERCENTILE_ACCURACY))
+ }
+
+ // Mark as lazy so that accuracyExpression is not evaluated during tree transformation.
+ private lazy val accuracy: Int = accuracyExpression.eval().asInstanceOf[Int]
+
+ private lazy val serializer: PercentileDigestSerializer = new PercentileDigestSerializer
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(DoubleType, TypeCollection(DoubleType, ArrayType), IntegerType)
+ }
+
+ // Mark as lazy so that percentageExpression is not evaluated during tree transformation.
+ private lazy val (returnPercentileArray: Boolean, percentages: Array[Double]) = {
+ (percentageExpression.dataType, percentageExpression.eval()) match {
+ // Rule ImplicitTypeCasts can cast other numeric types to double
+ case (_, num: Double) => (false, Array(num))
+ case (ArrayType(baseType: NumericType, _), arrayData: ArrayData) =>
+ val numericArray = arrayData.toObjectArray(baseType)
+ (true, numericArray.map { x =>
+ baseType.numeric.toDouble(x.asInstanceOf[baseType.InternalType])
+ })
+ case other =>
+ throw new AnalysisException(s"Invalid data type ${other._1} for parameter percentage")
+ }
+ }
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (accuracy <= 0) {
+ TypeCheckFailure(
+ s"The accuracy provided must be a positive integer literal (current value = $accuracy)")
+ } else if (percentages.exists(percentage => percentage < 0.0D || percentage > 1.0D)) {
+ TypeCheckFailure(
+ s"All percentage values must be between 0.0 and 1.0 " +
+ s"(current = ${percentages.mkString(", ")})")
+ } else {
+ TypeCheckSuccess
+ }
+ }
+
+ override def createAggregationBuffer(): PercentileDigest = {
+ val relativeError = 1.0D / accuracy
+ new PercentileDigest(relativeError)
+ }
+
+ override def update(buffer: PercentileDigest, inputRow: InternalRow): Unit = {
+ val value = child.eval(inputRow)
+ // Ignore empty rows, for example: percentile_approx(null)
+ if (value != null) {
+ buffer.add(value.asInstanceOf[Double])
+ }
+ }
+
+ override def merge(buffer: PercentileDigest, other: PercentileDigest): Unit = {
+ buffer.merge(other)
+ }
+
+ override def eval(buffer: PercentileDigest): Any = {
+ val result = buffer.getPercentiles(percentages)
+ if (result.length == 0) {
+ null
+ } else if (returnPercentileArray) {
+ new GenericArrayData(result)
+ } else {
+ result(0)
+ }
+ }
+
+ override def withNewMutableAggBufferOffset(newOffset: Int): ApproximatePercentile =
+ copy(mutableAggBufferOffset = newOffset)
+
+ override def withNewInputAggBufferOffset(newOffset: Int): ApproximatePercentile =
+ copy(inputAggBufferOffset = newOffset)
+
+ override def children: Seq[Expression] = Seq(child, percentageExpression, accuracyExpression)
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = {
+ if (returnPercentileArray) ArrayType(DoubleType) else DoubleType
+ }
+
+ override def prettyName: String = "percentile_approx"
+
+ override def serialize(obj: PercentileDigest): Array[Byte] = {
+ serializer.serialize(obj)
+ }
+
+ override def deserialize(bytes: Array[Byte]): PercentileDigest = {
+ serializer.deserialize(bytes)
+ }
+}
+
+object ApproximatePercentile {
+
+ // Default accuracy of Percentile approximation. Larger value means better accuracy.
+ // The default relative error can be deduced by defaultError = 1.0 / DEFAULT_PERCENTILE_ACCURACY
+ val DEFAULT_PERCENTILE_ACCURACY: Int = 10000
+
+ /**
+ * PercentileDigest is a probabilistic data structure used for approximating percentiles
+ * with limited memory. PercentileDigest is backed by [[QuantileSummaries]].
+ *
+ * @param summaries underlying probabilistic data structure [[QuantileSummaries]].
+ * @param isCompressed An internal flag from class [[QuantileSummaries]] to indicate whether the
+ * underlying quantileSummaries is compressed.
+ */
+ class PercentileDigest(
+ private var summaries: QuantileSummaries,
+ private var isCompressed: Boolean) {
+
+ // Trigger compression if the QuantileSummaries's buffer length exceeds
+ // compressThresHoldBufferLength. The buffer length can be get by
+ // quantileSummaries.sampled.length
+ private[this] final val compressThresHoldBufferLength: Int = {
+ // Max buffer length after compression.
+ val maxBufferLengthAfterCompression: Int = (1 / summaries.relativeError).toInt * 2
+ // A safe upper bound for buffer length before compression
+ maxBufferLengthAfterCompression * 2
+ }
+
+ def this(relativeError: Double) = {
+ this(new QuantileSummaries(defaultCompressThreshold, relativeError), isCompressed = true)
+ }
+
+ /** Returns compressed object of [[QuantileSummaries]] */
+ def quantileSummaries: QuantileSummaries = {
+ if (!isCompressed) compress()
+ summaries
+ }
+
+ /** Insert an observation value into the PercentileDigest data structure. */
+ def add(value: Double): Unit = {
+ summaries = summaries.insert(value)
+ // The result of QuantileSummaries.insert is un-compressed
+ isCompressed = false
+
+ // Currently, QuantileSummaries ignores the construction parameter compressThresHold,
+ // which may cause QuantileSummaries to occupy unbounded memory. We have to hack around here
+ // to make sure QuantileSummaries doesn't occupy infinite memory.
+ // TODO: Figure out why QuantileSummaries ignores construction parameter compressThresHold
+ if (summaries.sampled.length >= compressThresHoldBufferLength) compress()
+ }
+
+ /** In-place merges in another PercentileDigest. */
+ def merge(other: PercentileDigest): Unit = {
+ if (!isCompressed) compress()
+ summaries = summaries.merge(other.quantileSummaries)
+ }
+
+ /**
+ * Returns the approximate percentiles of all observation values at the given percentages.
+ * A percentile is a watermark value below which a given percentage of observation values fall.
+ * For example, the following code returns the 25th, median, and 75th percentiles of
+ * all observation values:
+ *
+ * {{{
+ * val Array(p25, median, p75) = percentileDigest.getPercentiles(Array(0.25, 0.5, 0.75))
+ * }}}
+ */
+ def getPercentiles(percentages: Array[Double]): Array[Double] = {
+ if (!isCompressed) compress()
--- End diff --
It is not supported by QuantileSummaries to evaluate an uncompressed QuantileSummaries
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14868: Implements percentile_approx aggregation function...
Posted by clockfly <gi...@git.apache.org>.
Github user clockfly commented on a diff in the pull request:
https://github.com/apache/spark/pull/14868#discussion_r76697179
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/ApproximatePercentileQuerySuite.scala ---
@@ -0,0 +1,226 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql
+
+import org.apache.spark.sql.catalyst.expressions.aggregate.ApproximatePercentile.DEFAULT_PERCENTILE_ACCURACY
+import org.apache.spark.sql.catalyst.expressions.aggregate.ApproximatePercentile.PercentileDigest
+import org.apache.spark.sql.test.SharedSQLContext
+
+class ApproximatePercentileQuerySuite extends QueryTest with SharedSQLContext {
+ import testImplicits._
+
+ private val table = "percentile_test"
+
+ test("percentile_approx, single percentile value") {
+ withTempView(table) {
+ (1 to 1000).toDF("col").createOrReplaceTempView(table)
+ checkAnswer(
+ spark.sql(
+ s"""
+ |SELECT
+ | percentile_approx(col, 0.25),
+ | percentile_approx(col, 0.5),
+ | percentile_approx(col, 0.75d),
+ | percentile_approx(col, 0.0),
+ | percentile_approx(col, 1.0),
+ | percentile_approx(col, 0),
+ | percentile_approx(col, 1)
+ |FROM $table
+ """.stripMargin),
+ Row(250D, 500D, 750D, 1D, 1000D, 1D, 1000D)
+ )
+ }
+ }
+
+ test("percentile_approx, array of percentile value") {
+ withTempView(table) {
+ (1 to 1000).toDF("col").createOrReplaceTempView(table)
+ checkAnswer(
+ spark.sql(
+ s"""SELECT
+ | percentile_approx(col, array(0.25, 0.5, 0.75D)),
+ | count(col),
+ | percentile_approx(col, array(0.0, 1.0)),
+ | sum(col)
+ |FROM $table
+ """.stripMargin),
+ Row(Seq(250D, 500D, 750D), 1000, Seq(1D, 1000D), 500500)
+ )
+ }
+ }
+
+ test("percentile_approx, with different accuracies") {
+
+ withTempView(table) {
+ (1 to 1000).toDF("col").createOrReplaceTempView(table)
+
+ // With different accuracies
+ val expectedPercentile = 250D
+ val accuracies = Array(1, 10, 100, 1000, 10000)
+ val errors = accuracies.map { accuracy =>
+ val df = spark.sql(s"SELECT percentile_approx(col, 0.25, $accuracy) FROM $table")
+ val approximatePercentile = df.collect().head.getDouble(0)
+ val error = Math.abs(approximatePercentile - expectedPercentile)
+ error
+ }
+
+ // The larger accuracy value we use, the smaller error we get
+ assert(errors.sorted.sameElements(errors.reverse))
+ }
+ }
+
+ test("percentile_approx, supports constant folding for parameter accuracy and percentages") {
+ withTempView(table) {
+ (1 to 1000).toDF("col").createOrReplaceTempView(table)
+ checkAnswer(
+ spark.sql(s"SELECT percentile_approx(col, array(0.25 + 0.25D), 200 + 800D) FROM $table"),
+ Row(Seq(500D))
+ )
+ }
+ }
+
+ test("percentile_approx(), aggregation on empty input table, no group by") {
+ withTempView(table) {
+ Seq.empty[Int].toDF("col").createOrReplaceTempView(table)
+ checkAnswer(
+ spark.sql(s"SELECT sum(col), percentile_approx(col, 0.5) FROM $table"),
+ Row(null, null)
+ )
+ }
+ }
+
+ test("percentile_approx(), aggregation on empty input table, with group by") {
--- End diff --
SortBasedAggregationIterator has different code path for "with group by" and "no group by", so it is better that we test both cases.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: [SPARK-16283][SQL][WIP] Implements percentile_approx agg...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14868
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/64676/
Test PASSed.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: [SPARK-16283][SQL][WIP] Implements percentile_approx agg...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14868
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/64687/
Test PASSed.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: Implements percentile_approx aggregation function which ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14868
Merged build finished. Test FAILed.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: [SPARK-16283][SQL][WIP] Implements percentile_approx agg...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14868
**[Test build #64695 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64695/consoleFull)** for PR 14868 at commit [`bc70a00`](https://github.com/apache/spark/commit/bc70a0023bb24175c06c03cb7acad7f9a6d34e36).
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14868: Implements percentile_approx aggregation function...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/14868#discussion_r76781601
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/ApproximatePercentile.scala ---
@@ -0,0 +1,304 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions.aggregate
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.{InternalRow, ScalaReflectionLock}
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.{TypeCheckFailure, TypeCheckSuccess}
+import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.ApproximatePercentile.{PercentileDigest, PercentileDigestSerializer}
+import org.apache.spark.sql.catalyst.util.{ArrayData, GenericArrayData}
+import org.apache.spark.sql.catalyst.util.QuantileSummaries
+import org.apache.spark.sql.catalyst.util.QuantileSummaries.{defaultCompressThreshold, Stats}
+import org.apache.spark.sql.types._
+
+/**
+ * The ApproximatePercentile function returns the approximate percentile(s) of a column at the given
+ * percentage(s). A percentile is a watermark value below which a given percentage of the column
+ * values fall. For example, the percentile of column `col` at percentage 50% is the median of
+ * column `col`.
+ *
+ * This function supports partial aggregation.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param percentageExpression Expression that represents a single percentage value or
+ * a array of percentage values. Each percentage value must be between
+ * 0.0 and 1.0.
+ * @param accuracyExpression Integer literal expression of approximation accuracy. Higher value
+ * yields better accuracy, the default value is
+ * DEFAULT_PERCENTILE_ACCURACY.
+ */
+@ExpressionDescription(
+ usage =
+ """
+ _FUNC_(col, percentage [, accuracy]) - Returns the approximate percentile value of numeric
+ column `col` at the given percentage. The value of percentage must be between 0.0
+ and 1.0. The `accuracy` parameter (default: 10000) is a positive integer literal which
+ controls approximation accuracy at the cost of memory. Higher value of `accuracy` yields
+ better accuracy, `1.0/accuracy` is the relative error of the approximation.
+
+ _FUNC_(col, array(percentage1 [, percentage2]...) [, accuracy]) - Returns the approximate
+ percentile array of column `col` at the given percentage array. Each value of the
+ percentage array must be between 0.0 and 1.0. The `accuracy` parameter (default: 10000) is
+ a positive integer literal which controls approximation accuracy at the cost of memory.
+ Higher value of `accuracy` yields better accuracy, `1.0/accuracy` is the relative error of
+ the approximation.
+ """)
+case class ApproximatePercentile(
+ child: Expression,
+ percentageExpression: Expression,
--- End diff --
why not follow this style? https://github.com/apache/spark/blob/master/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/HyperLogLogPlusPlus.scala#L56
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: Implements percentile_approx aggregation function which ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14868
**[Test build #64687 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64687/consoleFull)** for PR 14868 at commit [`421e211`](https://github.com/apache/spark/commit/421e2118d3883ff57b01704327f887eb6d6ea3e8).
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: [SPARK-16283][SQL] Implements percentile_approx aggregat...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on the issue:
https://github.com/apache/spark/pull/14868
LGTM, merging to master!
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: Implements percentile_approx aggregation function which ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14868
**[Test build #64653 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64653/consoleFull)** for PR 14868 at commit [`539d693`](https://github.com/apache/spark/commit/539d693d7a569cc83019526ea2565de5a86fda83).
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: Implements percentile_approx aggregation function which ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14868
**[Test build #64653 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64653/consoleFull)** for PR 14868 at commit [`539d693`](https://github.com/apache/spark/commit/539d693d7a569cc83019526ea2565de5a86fda83).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: [SPARK-16283][SQL][WIP] Implements percentile_approx agg...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14868
Merged build finished. Test PASSed.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: Implements percentile_approx aggregation function which ...
Posted by liancheng <gi...@git.apache.org>.
Github user liancheng commented on the issue:
https://github.com/apache/spark/pull/14868
LGTM except for a few minor issues. The failed test should be a typo in [this line].
[1]: https://github.com/apache/spark/pull/14868/files#diff-9b242f568b320219b1e2cfa2ee10137bR173
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: Implements percentile_approx aggregation function which ...
Posted by clockfly <gi...@git.apache.org>.
Github user clockfly commented on the issue:
https://github.com/apache/spark/pull/14868
@yhuai @cloud-fan, can you take a look?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14868: [SPARK-16283][SQL] Implements percentile_approx a...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/14868#discussion_r77114139
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/ApproximatePercentile.scala ---
@@ -0,0 +1,321 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions.aggregate
+
+import java.nio.ByteBuffer
+
+import com.google.common.primitives.{Doubles, Ints, Longs}
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.{InternalRow}
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.{TypeCheckFailure, TypeCheckSuccess}
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.ApproximatePercentile.{PercentileDigest}
+import org.apache.spark.sql.catalyst.util.{ArrayData, GenericArrayData}
+import org.apache.spark.sql.catalyst.util.QuantileSummaries
+import org.apache.spark.sql.catalyst.util.QuantileSummaries.{defaultCompressThreshold, Stats}
+import org.apache.spark.sql.types._
+
+/**
+ * The ApproximatePercentile function returns the approximate percentile(s) of a column at the given
+ * percentage(s). A percentile is a watermark value below which a given percentage of the column
+ * values fall. For example, the percentile of column `col` at percentage 50% is the median of
+ * column `col`.
+ *
+ * This function supports partial aggregation.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param percentageExpression Expression that represents a single percentage value or
+ * an array of percentage values. Each percentage value must be between
+ * 0.0 and 1.0.
+ * @param accuracyExpression Integer literal expression of approximation accuracy. Higher value
+ * yields better accuracy, the default value is
+ * DEFAULT_PERCENTILE_ACCURACY.
+ */
+@ExpressionDescription(
+ usage =
+ """
+ _FUNC_(col, percentage [, accuracy]) - Returns the approximate percentile value of numeric
+ column `col` at the given percentage. The value of percentage must be between 0.0
+ and 1.0. The `accuracy` parameter (default: 10000) is a positive integer literal which
+ controls approximation accuracy at the cost of memory. Higher value of `accuracy` yields
+ better accuracy, `1.0/accuracy` is the relative error of the approximation.
+
+ _FUNC_(col, array(percentage1 [, percentage2]...) [, accuracy]) - Returns the approximate
+ percentile array of column `col` at the given percentage array. Each value of the
+ percentage array must be between 0.0 and 1.0. The `accuracy` parameter (default: 10000) is
+ a positive integer literal which controls approximation accuracy at the cost of memory.
+ Higher value of `accuracy` yields better accuracy, `1.0/accuracy` is the relative error of
+ the approximation.
+ """)
+case class ApproximatePercentile(
+ child: Expression,
+ percentageExpression: Expression,
+ accuracyExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int) extends TypedImperativeAggregate[PercentileDigest] {
+
+ def this(child: Expression, percentageExpression: Expression, accuracyExpression: Expression) = {
+ this(child, percentageExpression, accuracyExpression, 0, 0)
+ }
+
+ def this(child: Expression, percentageExpression: Expression) = {
+ this(child, percentageExpression, Literal(ApproximatePercentile.DEFAULT_PERCENTILE_ACCURACY))
+ }
+
+ // Mark as lazy so that accuracyExpression is not evaluated during tree transformation.
+ private lazy val accuracy: Int = accuracyExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(DoubleType, TypeCollection(DoubleType, ArrayType), IntegerType)
+ }
+
+ // Mark as lazy so that percentageExpression is not evaluated during tree transformation.
+ private lazy val (returnPercentileArray: Boolean, percentages: Array[Double]) = {
+ (percentageExpression.dataType, percentageExpression.eval()) match {
+ // Rule ImplicitTypeCasts can cast other numeric types to double
+ case (_, num: Double) => (false, Array(num))
+ case (ArrayType(baseType: NumericType, _), arrayData: ArrayData) =>
+ val numericArray = arrayData.toObjectArray(baseType)
+ (true, numericArray.map { x =>
+ baseType.numeric.toDouble(x.asInstanceOf[baseType.InternalType])
+ })
+ case other =>
+ throw new AnalysisException(s"Invalid data type ${other._1} for parameter percentage")
+ }
+ }
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (!percentageExpression.foldable || !accuracyExpression.foldable) {
+ TypeCheckFailure(s"The accuracy or percentage provided must be a constant literal")
+ } else if (accuracy <= 0) {
+ TypeCheckFailure(
+ s"The accuracy provided must be a positive integer literal (current value = $accuracy)")
+ } else if (percentages.exists(percentage => percentage < 0.0D || percentage > 1.0D)) {
+ TypeCheckFailure(
+ s"All percentage values must be between 0.0 and 1.0 " +
+ s"(current = ${percentages.mkString(", ")})")
+ } else {
+ TypeCheckSuccess
+ }
+ }
+
+ override def createAggregationBuffer(): PercentileDigest = {
+ val relativeError = 1.0D / accuracy
+ new PercentileDigest(relativeError)
+ }
+
+ override def update(buffer: PercentileDigest, inputRow: InternalRow): Unit = {
+ val value = child.eval(inputRow)
+ // Ignore empty rows, for example: percentile_approx(null)
+ if (value != null) {
+ buffer.add(value.asInstanceOf[Double])
+ }
+ }
+
+ override def merge(buffer: PercentileDigest, other: PercentileDigest): Unit = {
+ buffer.merge(other)
+ }
+
+ override def eval(buffer: PercentileDigest): Any = {
+ val result = buffer.getPercentiles(percentages)
+ if (result.length == 0) {
+ null
+ } else if (returnPercentileArray) {
+ new GenericArrayData(result)
+ } else {
+ result(0)
+ }
+ }
+
+ override def withNewMutableAggBufferOffset(newOffset: Int): ApproximatePercentile =
+ copy(mutableAggBufferOffset = newOffset)
+
+ override def withNewInputAggBufferOffset(newOffset: Int): ApproximatePercentile =
+ copy(inputAggBufferOffset = newOffset)
+
+ override def children: Seq[Expression] = Seq(child, percentageExpression, accuracyExpression)
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = {
+ if (returnPercentileArray) ArrayType(DoubleType) else DoubleType
+ }
+
+ override def prettyName: String = "percentile_approx"
+
+ override def serialize(obj: PercentileDigest): Array[Byte] = {
+ ApproximatePercentile.serializer.serialize(obj)
+ }
+
+ override def deserialize(bytes: Array[Byte]): PercentileDigest = {
+ ApproximatePercentile.serializer.deserialize(bytes)
+ }
+}
+
+object ApproximatePercentile {
+
+ // Default accuracy of Percentile approximation. Larger value means better accuracy.
+ // The default relative error can be deduced by defaultError = 1.0 / DEFAULT_PERCENTILE_ACCURACY
+ val DEFAULT_PERCENTILE_ACCURACY: Int = 10000
+
+ /**
+ * PercentileDigest is a probabilistic data structure used for approximating percentiles
+ * with limited memory. PercentileDigest is backed by [[QuantileSummaries]].
+ *
+ * @param summaries underlying probabilistic data structure [[QuantileSummaries]].
+ * @param isCompressed An internal flag from class [[QuantileSummaries]] to indicate whether the
+ * underlying quantileSummaries is compressed.
+ */
+ class PercentileDigest(
+ private var summaries: QuantileSummaries,
+ private var isCompressed: Boolean) {
+
+ // Trigger compression if the QuantileSummaries's buffer length exceeds
+ // compressThresHoldBufferLength. The buffer length can be get by
+ // quantileSummaries.sampled.length
+ private[this] final val compressThresHoldBufferLength: Int = {
+ // Max buffer length after compression.
+ val maxBufferLengthAfterCompression: Int = (1 / summaries.relativeError).toInt * 2
+ // A safe upper bound for buffer length before compression
+ maxBufferLengthAfterCompression * 2
+ }
+
+ def this(relativeError: Double) = {
+ this(new QuantileSummaries(defaultCompressThreshold, relativeError), isCompressed = true)
+ }
+
+ /** Returns compressed object of [[QuantileSummaries]] */
+ def quantileSummaries: QuantileSummaries = {
+ if (!isCompressed) compress()
+ summaries
+ }
+
+ /** Insert an observation value into the PercentileDigest data structure. */
+ def add(value: Double): Unit = {
+ summaries = summaries.insert(value)
+ // The result of QuantileSummaries.insert is un-compressed
+ isCompressed = false
+
+ // Currently, QuantileSummaries ignores the construction parameter compressThresHold,
+ // which may cause QuantileSummaries to occupy unbounded memory. We have to hack around here
+ // to make sure QuantileSummaries doesn't occupy infinite memory.
+ // TODO: Figure out why QuantileSummaries ignores construction parameter compressThresHold
+ if (summaries.sampled.length >= compressThresHoldBufferLength) compress()
+ }
+
+ /** In-place merges in another PercentileDigest. */
+ def merge(other: PercentileDigest): Unit = {
+ if (!isCompressed) compress()
+ summaries = summaries.merge(other.quantileSummaries)
--- End diff --
does the `other` need be compressed too?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: Implements percentile_approx aggregation function which ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14868
**[Test build #64621 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64621/consoleFull)** for PR 14868 at commit [`dc09d8c`](https://github.com/apache/spark/commit/dc09d8c2f6191c98fefe4b9dfe7bdd817907560c).
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14868: [SPARK-16283][SQL] Implements percentile_approx a...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/14868#discussion_r77114814
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/ApproximatePercentile.scala ---
@@ -0,0 +1,321 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions.aggregate
+
+import java.nio.ByteBuffer
+
+import com.google.common.primitives.{Doubles, Ints, Longs}
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.{InternalRow}
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.{TypeCheckFailure, TypeCheckSuccess}
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.ApproximatePercentile.{PercentileDigest}
+import org.apache.spark.sql.catalyst.util.{ArrayData, GenericArrayData}
+import org.apache.spark.sql.catalyst.util.QuantileSummaries
+import org.apache.spark.sql.catalyst.util.QuantileSummaries.{defaultCompressThreshold, Stats}
+import org.apache.spark.sql.types._
+
+/**
+ * The ApproximatePercentile function returns the approximate percentile(s) of a column at the given
+ * percentage(s). A percentile is a watermark value below which a given percentage of the column
+ * values fall. For example, the percentile of column `col` at percentage 50% is the median of
+ * column `col`.
+ *
+ * This function supports partial aggregation.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param percentageExpression Expression that represents a single percentage value or
+ * an array of percentage values. Each percentage value must be between
+ * 0.0 and 1.0.
+ * @param accuracyExpression Integer literal expression of approximation accuracy. Higher value
+ * yields better accuracy, the default value is
+ * DEFAULT_PERCENTILE_ACCURACY.
+ */
+@ExpressionDescription(
+ usage =
+ """
+ _FUNC_(col, percentage [, accuracy]) - Returns the approximate percentile value of numeric
+ column `col` at the given percentage. The value of percentage must be between 0.0
+ and 1.0. The `accuracy` parameter (default: 10000) is a positive integer literal which
+ controls approximation accuracy at the cost of memory. Higher value of `accuracy` yields
+ better accuracy, `1.0/accuracy` is the relative error of the approximation.
+
+ _FUNC_(col, array(percentage1 [, percentage2]...) [, accuracy]) - Returns the approximate
+ percentile array of column `col` at the given percentage array. Each value of the
+ percentage array must be between 0.0 and 1.0. The `accuracy` parameter (default: 10000) is
+ a positive integer literal which controls approximation accuracy at the cost of memory.
+ Higher value of `accuracy` yields better accuracy, `1.0/accuracy` is the relative error of
+ the approximation.
+ """)
+case class ApproximatePercentile(
+ child: Expression,
+ percentageExpression: Expression,
+ accuracyExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int) extends TypedImperativeAggregate[PercentileDigest] {
+
+ def this(child: Expression, percentageExpression: Expression, accuracyExpression: Expression) = {
+ this(child, percentageExpression, accuracyExpression, 0, 0)
+ }
+
+ def this(child: Expression, percentageExpression: Expression) = {
+ this(child, percentageExpression, Literal(ApproximatePercentile.DEFAULT_PERCENTILE_ACCURACY))
+ }
+
+ // Mark as lazy so that accuracyExpression is not evaluated during tree transformation.
+ private lazy val accuracy: Int = accuracyExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(DoubleType, TypeCollection(DoubleType, ArrayType), IntegerType)
+ }
+
+ // Mark as lazy so that percentageExpression is not evaluated during tree transformation.
+ private lazy val (returnPercentileArray: Boolean, percentages: Array[Double]) = {
+ (percentageExpression.dataType, percentageExpression.eval()) match {
+ // Rule ImplicitTypeCasts can cast other numeric types to double
+ case (_, num: Double) => (false, Array(num))
+ case (ArrayType(baseType: NumericType, _), arrayData: ArrayData) =>
+ val numericArray = arrayData.toObjectArray(baseType)
+ (true, numericArray.map { x =>
+ baseType.numeric.toDouble(x.asInstanceOf[baseType.InternalType])
+ })
+ case other =>
+ throw new AnalysisException(s"Invalid data type ${other._1} for parameter percentage")
+ }
+ }
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (!percentageExpression.foldable || !accuracyExpression.foldable) {
+ TypeCheckFailure(s"The accuracy or percentage provided must be a constant literal")
+ } else if (accuracy <= 0) {
+ TypeCheckFailure(
+ s"The accuracy provided must be a positive integer literal (current value = $accuracy)")
+ } else if (percentages.exists(percentage => percentage < 0.0D || percentage > 1.0D)) {
+ TypeCheckFailure(
+ s"All percentage values must be between 0.0 and 1.0 " +
+ s"(current = ${percentages.mkString(", ")})")
+ } else {
+ TypeCheckSuccess
+ }
+ }
+
+ override def createAggregationBuffer(): PercentileDigest = {
+ val relativeError = 1.0D / accuracy
+ new PercentileDigest(relativeError)
+ }
+
+ override def update(buffer: PercentileDigest, inputRow: InternalRow): Unit = {
+ val value = child.eval(inputRow)
+ // Ignore empty rows, for example: percentile_approx(null)
+ if (value != null) {
+ buffer.add(value.asInstanceOf[Double])
+ }
+ }
+
+ override def merge(buffer: PercentileDigest, other: PercentileDigest): Unit = {
+ buffer.merge(other)
+ }
+
+ override def eval(buffer: PercentileDigest): Any = {
+ val result = buffer.getPercentiles(percentages)
+ if (result.length == 0) {
+ null
+ } else if (returnPercentileArray) {
+ new GenericArrayData(result)
+ } else {
+ result(0)
+ }
+ }
+
+ override def withNewMutableAggBufferOffset(newOffset: Int): ApproximatePercentile =
+ copy(mutableAggBufferOffset = newOffset)
+
+ override def withNewInputAggBufferOffset(newOffset: Int): ApproximatePercentile =
+ copy(inputAggBufferOffset = newOffset)
+
+ override def children: Seq[Expression] = Seq(child, percentageExpression, accuracyExpression)
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = {
+ if (returnPercentileArray) ArrayType(DoubleType) else DoubleType
+ }
+
+ override def prettyName: String = "percentile_approx"
+
+ override def serialize(obj: PercentileDigest): Array[Byte] = {
+ ApproximatePercentile.serializer.serialize(obj)
+ }
+
+ override def deserialize(bytes: Array[Byte]): PercentileDigest = {
+ ApproximatePercentile.serializer.deserialize(bytes)
+ }
+}
+
+object ApproximatePercentile {
+
+ // Default accuracy of Percentile approximation. Larger value means better accuracy.
+ // The default relative error can be deduced by defaultError = 1.0 / DEFAULT_PERCENTILE_ACCURACY
+ val DEFAULT_PERCENTILE_ACCURACY: Int = 10000
+
+ /**
+ * PercentileDigest is a probabilistic data structure used for approximating percentiles
+ * with limited memory. PercentileDigest is backed by [[QuantileSummaries]].
+ *
+ * @param summaries underlying probabilistic data structure [[QuantileSummaries]].
+ * @param isCompressed An internal flag from class [[QuantileSummaries]] to indicate whether the
+ * underlying quantileSummaries is compressed.
+ */
+ class PercentileDigest(
+ private var summaries: QuantileSummaries,
+ private var isCompressed: Boolean) {
--- End diff --
why we need this flag? `QuantileSummaries` provides a way to show it's compressed or not: `headSampled.isEmpty`
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14868: Implements percentile_approx aggregation function...
Posted by liancheng <gi...@git.apache.org>.
Github user liancheng commented on a diff in the pull request:
https://github.com/apache/spark/pull/14868#discussion_r76761692
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/ApproximatePercentile.scala ---
@@ -0,0 +1,304 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions.aggregate
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.{InternalRow, ScalaReflectionLock}
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.{TypeCheckFailure, TypeCheckSuccess}
+import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.ApproximatePercentile.{PercentileDigest, PercentileDigestSerializer}
+import org.apache.spark.sql.catalyst.util.{ArrayData, GenericArrayData}
+import org.apache.spark.sql.catalyst.util.QuantileSummaries
+import org.apache.spark.sql.catalyst.util.QuantileSummaries.{defaultCompressThreshold, Stats}
+import org.apache.spark.sql.types._
+
+/**
+ * The ApproximatePercentile function returns the approximate percentile(s) of a column at the given
+ * percentage(s). A percentile is a watermark value below which a given percentage of the column
+ * values fall. For example, the percentile of column `col` at percentage 50% is the median of
+ * column `col`.
+ *
+ * This function supports partial aggregation.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param percentageExpression Expression that represents a single percentage value or
+ * a array of percentage values. Each percentage value must be between
+ * 0.0 and 1.0.
+ * @param accuracyExpression Integer literal expression of approximation accuracy. Higher value
+ * yield better accuracy, the default value is
+ * DEFAULT_PERCENTILE_ACCURACY.
+ */
+@ExpressionDescription(
+ usage =
+ """
+ _FUNC_(col, percentage [, accuracy]) - Returns the approximate percentile value of numeric
+ column `col` at the given percentage. The value of percentage must be between 0.0
+ and 1.0. The `accuracy` parameter (default: 10000) is a positive integer literal which
+ controls approximation accuracy at the cost of memory. Higher value of `accuracy` yield
--- End diff --
Nit: yield => yields
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: Implements percentile_approx aggregation function which ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14868
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/64621/
Test FAILed.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: Implements percentile_approx aggregation function which ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14868
**[Test build #64623 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64623/consoleFull)** for PR 14868 at commit [`9d7fb85`](https://github.com/apache/spark/commit/9d7fb85f6f2741f050344b06f2efac3254da77bf).
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: Implements percentile_approx aggregation function which ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14868
**[Test build #64639 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64639/consoleFull)** for PR 14868 at commit [`9d7fb85`](https://github.com/apache/spark/commit/9d7fb85f6f2741f050344b06f2efac3254da77bf).
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: Implements percentile_approx aggregation function which ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14868
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/64653/
Test FAILed.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: Implements percentile_approx aggregation function which ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14868
**[Test build #64643 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64643/consoleFull)** for PR 14868 at commit [`3dda1ae`](https://github.com/apache/spark/commit/3dda1ae90cea0eab8dd963d4af5708bff8846d65).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: Implements percentile_approx aggregation function which ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14868
**[Test build #64600 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64600/consoleFull)** for PR 14868 at commit [`88232ed`](https://github.com/apache/spark/commit/88232ed07780be3243c18ba1d3e7943cda628dcc).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds the following public classes _(experimental)_:
* `case class ApproximatePercentile(`
* ` class PercentileDigest(`
* ` class PercentileDigestSerializer `
* `case class QuantileSummaries(`
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14868: Implements percentile_approx aggregation function...
Posted by clockfly <gi...@git.apache.org>.
Github user clockfly commented on a diff in the pull request:
https://github.com/apache/spark/pull/14868#discussion_r76695715
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/ApproximatePercentile.scala ---
@@ -0,0 +1,302 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions.aggregate
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.{InternalRow, ScalaReflectionLock}
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.{TypeCheckFailure, TypeCheckSuccess}
+import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.ApproximatePercentile.{PercentileDigest, PercentileDigestSerializer}
+import org.apache.spark.sql.catalyst.util.{ArrayData, GenericArrayData}
+import org.apache.spark.sql.catalyst.util.QuantileSummaries
+import org.apache.spark.sql.catalyst.util.QuantileSummaries.{defaultCompressThreshold, Stats}
+import org.apache.spark.sql.types._
+
+/**
+ * The ApproximatePercentile function returns the approximate percentile(s) of a column at the given
+ * percentage(s). A percentile is a watermark value below which a given percentage of the column
+ * values fall. For example, the percentile of column `col` at percentage 50% is the median of
+ * column `col`.
+ *
+ * This function supports partial aggregation.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param percentageExpression Expression that represents a single percentage value or
+ * a array of percentage values. Each percentage value must be between
+ * 0.0 and 1.0.
+ * @param accuracyExpression Integer literal expression of approximation accuracy. Higher value
+ * yield better accuracy, the default value is
+ * DEFAULT_PERCENTILE_ACCURACY.
+ */
+@ExpressionDescription(
+ usage =
+ """
+ _FUNC_(col, percentage [, accuracy]) - Returns the approximate percentile value of numeric
+ column `col` at the given percentage. The value of percentage must be between 0.0
+ and 1.0. The `accuracy` parameter (default: 10000) is a positive integer literal which
+ controls approximation accuracy at the cost of memory. Higher value of `accuracy` yield
+ better accuracy, `1.0/accuracy` is the relative error of the approximation.
+
+ _FUNC_(col, array(percentage1 [, percentage2]...) [, accuracy]) - Returns the approximate
+ percentile array of column `col` at the given percentage array. Each value of the
+ percentage array must be between 0.0 and 1.0. The `accuracy` parameter (default: 10000) is
+ a positive integer literal which controls approximation accuracy at the cost of memory.
+ Higher value of `accuracy` yield better accuracy, `1.0/accuracy` is the relative error of
+ the approximation.
+ """)
+case class ApproximatePercentile(
+ child: Expression,
+ percentageExpression: Expression,
+ accuracyExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int) extends TypedImperativeAggregate[PercentileDigest] {
+
+ def this(child: Expression, percentageExpression: Expression, accuracyExpression: Expression) = {
+ this(child, percentageExpression, accuracyExpression, 0, 0)
+ }
+
+ def this(child: Expression, percentageExpression: Expression) = {
+ this(child, percentageExpression, Literal(ApproximatePercentile.DEFAULT_PERCENTILE_ACCURACY))
+ }
+
+ // Mark as lazy so that accuracyExpression is not evaluated during tree transformation.
+ private lazy val accuracy: Int = accuracyExpression.eval().asInstanceOf[Int]
+
+ private lazy val serializer: PercentileDigestSerializer = new PercentileDigestSerializer
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(DoubleType, TypeCollection(DoubleType, ArrayType), IntegerType)
+ }
+
+ // Mark as lazy so that percentageExpression is not evaluated during tree transformation.
+ private lazy val (returnPercentileArray: Boolean, percentages: Array[Double]) = {
+ (percentageExpression.dataType, percentageExpression.eval()) match {
+ // Rule ImplicitTypeCasts can cast other numeric types to double
+ case (_, num: Double) => (false, Array(num))
+ case (ArrayType(baseType: NumericType, _), arrayData: ArrayData) =>
+ val numericArray = arrayData.toArray(baseType)(baseType.classTag)
+ (true, numericArray.map(baseType.numeric.toDouble))
+ case other =>
+ throw new AnalysisException(s"Invalid data type ${other._1} for parameter percentage")
+ }
+ }
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (accuracy <= 0) {
+ TypeCheckFailure(
+ s"The accuracy provided must be a positive integer literal (current value = $accuracy)")
+ } else if (percentages.exists(percentage => percentage < 0.0D || percentage > 1.0D)) {
+ TypeCheckFailure(
+ s"All percentage values must be between 0.0 and 1.0 " +
+ s"(current = ${percentages.mkString(", ")})")
+ } else {
+ TypeCheckSuccess
+ }
+ }
+
+ override def createAggregationBuffer(): PercentileDigest = {
+ val relativeError = 1.0D / accuracy
+ new PercentileDigest(relativeError)
+ }
+
+ override def update(buffer: PercentileDigest, inputRow: InternalRow): Unit = {
+ val value = child.eval(inputRow)
+ // Ignore empty rows, for example: percentile_approx(null)
+ if (value != null) {
+ buffer.add(value.asInstanceOf[Double])
+ }
+ }
+
+ override def merge(buffer: PercentileDigest, other: PercentileDigest): Unit = {
+ buffer.merge(other)
+ }
+
+ override def eval(buffer: PercentileDigest): Any = {
+ val result = buffer.getPercentiles(percentages)
+ if (result.length == 0) {
+ null
+ } else if (returnPercentileArray) {
+ new GenericArrayData(result)
+ } else {
+ result(0)
+ }
+ }
+
+ override def withNewMutableAggBufferOffset(newOffset: Int): ApproximatePercentile =
+ copy(mutableAggBufferOffset = newOffset)
+
+ override def withNewInputAggBufferOffset(newOffset: Int): ApproximatePercentile =
+ copy(inputAggBufferOffset = newOffset)
+
+ override def children: Seq[Expression] = Seq(child, percentageExpression, accuracyExpression)
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = {
+ if (returnPercentileArray) ArrayType(DoubleType) else DoubleType
+ }
+
+ override def prettyName: String = "percentile_approx"
+
+ override def serialize(obj: PercentileDigest): Array[Byte] = {
+ serializer.serialize(obj)
+ }
+
+ override def deserialize(bytes: Array[Byte]): PercentileDigest = {
+ serializer.deserialize(bytes)
+ }
+}
+
+object ApproximatePercentile {
+
+ // Default accuracy of Percentile approximation. Larger value means better accuracy.
+ // The default relative error can be deduced by defaultError = 1.0 / DEFAULT_PERCENTILE_ACCURACY
+ val DEFAULT_PERCENTILE_ACCURACY: Int = 10000
+
+ /**
+ * PercentileDigest is a probabilistic data structure used for approximating percentiles
+ * with limited memory. PercentileDigest is backed by [[QuantileSummaries]].
+ *
+ * @param summaries underlying probabilistic data structure [[QuantileSummaries]].
+ * @param isCompressed An internal flag from class [[QuantileSummaries]] to indicate whether the
+ * underlying quantileSummaries is compressed.
+ */
+ class PercentileDigest(
+ private var summaries: QuantileSummaries,
+ private var isCompressed: Boolean) {
+
+ // Trigger compression if the QuantileSummaries's buffer length exceeds
+ // compressThresHoldBufferLength. The buffer length can be get by
+ // quantileSummaries.sampled.length
+ private[this] final val compressThresHoldBufferLength: Int = {
+ // Max buffer length after compression.
+ val maxBufferLengthAfterCompression: Int = (1 / summaries.relativeError).toInt * 2
+ // A safe upper bound for buffer length before compression
+ maxBufferLengthAfterCompression * 2
+ }
+
+ def this(relativeError: Double) = {
+ this(new QuantileSummaries(defaultCompressThreshold, relativeError), isCompressed = true)
+ }
+
+ /** Returns compressed object of [[QuantileSummaries]] */
+ def quantileSummaries: QuantileSummaries = {
+ if (!isCompressed) compress()
+ summaries
+ }
+
+ /** Insert an observation value into the PercentileDigest data structure. */
+ def add(value: Double): Unit = {
+ summaries = summaries.insert(value)
+ // The result of QuantileSummaries.insert is un-compressed
+ isCompressed = false
+
+ // Currently, QuantileSummaries ignores the construction parameter compressThresHold,
+ // which may cause QuantileSummaries to occupy unbounded memory. We have to hack around here
+ // to make sure QuantileSummaries doesn't occupy infinite memory.
+ // TODO: Figure out why QuantileSummaries ignores construction parameter compressThresHold
+ if (summaries.sampled.length >= compressThresHoldBufferLength) compress()
--- End diff --
It is surprising to see QuantileSummaries doesn't do automatic compression. Need to check with @thunterdb
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: Implements percentile_approx aggregation function which ...
Posted by clockfly <gi...@git.apache.org>.
Github user clockfly commented on the issue:
https://github.com/apache/spark/pull/14868
@thunterdb
1. Looks like current impl of QuantileSummaries will not do automatically compression in some cases.
For example:
`(1 to 10000000).foreach(qualileSummaries.insert(_))` probably will trigger OOM. Is this expected?
2. The following code will add two sample Stats in QuantileSummaries, is this expected? This results in some strange behavior when used in window function.
```
val quantileSummaries = new QuantileSummaries(1000, 0.001)
quantileSummaries.insert(100.0D)
val updated = quantileSummaries.compress()
// Is this supposed to be 2?
assert(updated.sampled.length == 2)
```
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14868: Implements percentile_approx aggregation function...
Posted by clockfly <gi...@git.apache.org>.
Github user clockfly commented on a diff in the pull request:
https://github.com/apache/spark/pull/14868#discussion_r76898106
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/ApproximatePercentile.scala ---
@@ -0,0 +1,304 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions.aggregate
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.{InternalRow, ScalaReflectionLock}
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.{TypeCheckFailure, TypeCheckSuccess}
+import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.ApproximatePercentile.{PercentileDigest, PercentileDigestSerializer}
+import org.apache.spark.sql.catalyst.util.{ArrayData, GenericArrayData}
+import org.apache.spark.sql.catalyst.util.QuantileSummaries
+import org.apache.spark.sql.catalyst.util.QuantileSummaries.{defaultCompressThreshold, Stats}
+import org.apache.spark.sql.types._
+
+/**
+ * The ApproximatePercentile function returns the approximate percentile(s) of a column at the given
+ * percentage(s). A percentile is a watermark value below which a given percentage of the column
+ * values fall. For example, the percentile of column `col` at percentage 50% is the median of
+ * column `col`.
+ *
+ * This function supports partial aggregation.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param percentageExpression Expression that represents a single percentage value or
+ * a array of percentage values. Each percentage value must be between
+ * 0.0 and 1.0.
+ * @param accuracyExpression Integer literal expression of approximation accuracy. Higher value
+ * yields better accuracy, the default value is
+ * DEFAULT_PERCENTILE_ACCURACY.
+ */
+@ExpressionDescription(
+ usage =
+ """
+ _FUNC_(col, percentage [, accuracy]) - Returns the approximate percentile value of numeric
+ column `col` at the given percentage. The value of percentage must be between 0.0
+ and 1.0. The `accuracy` parameter (default: 10000) is a positive integer literal which
+ controls approximation accuracy at the cost of memory. Higher value of `accuracy` yields
+ better accuracy, `1.0/accuracy` is the relative error of the approximation.
+
+ _FUNC_(col, array(percentage1 [, percentage2]...) [, accuracy]) - Returns the approximate
+ percentile array of column `col` at the given percentage array. Each value of the
+ percentage array must be between 0.0 and 1.0. The `accuracy` parameter (default: 10000) is
+ a positive integer literal which controls approximation accuracy at the cost of memory.
+ Higher value of `accuracy` yields better accuracy, `1.0/accuracy` is the relative error of
+ the approximation.
+ """)
+case class ApproximatePercentile(
+ child: Expression,
+ percentageExpression: Expression,
--- End diff --
The sql string is automatically generated, without hacking.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: [SPARK-16283][SQL][WIP] Implements percentile_approx agg...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14868
Merged build finished. Test PASSed.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14868: [SPARK-16283][SQL] Implements percentile_approx a...
Posted by clockfly <gi...@git.apache.org>.
Github user clockfly commented on a diff in the pull request:
https://github.com/apache/spark/pull/14868#discussion_r77126967
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/ApproximatePercentile.scala ---
@@ -0,0 +1,321 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions.aggregate
+
+import java.nio.ByteBuffer
+
+import com.google.common.primitives.{Doubles, Ints, Longs}
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.{InternalRow}
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.{TypeCheckFailure, TypeCheckSuccess}
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.ApproximatePercentile.{PercentileDigest}
+import org.apache.spark.sql.catalyst.util.{ArrayData, GenericArrayData}
+import org.apache.spark.sql.catalyst.util.QuantileSummaries
+import org.apache.spark.sql.catalyst.util.QuantileSummaries.{defaultCompressThreshold, Stats}
+import org.apache.spark.sql.types._
+
+/**
+ * The ApproximatePercentile function returns the approximate percentile(s) of a column at the given
+ * percentage(s). A percentile is a watermark value below which a given percentage of the column
+ * values fall. For example, the percentile of column `col` at percentage 50% is the median of
+ * column `col`.
+ *
+ * This function supports partial aggregation.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param percentageExpression Expression that represents a single percentage value or
+ * an array of percentage values. Each percentage value must be between
+ * 0.0 and 1.0.
+ * @param accuracyExpression Integer literal expression of approximation accuracy. Higher value
+ * yields better accuracy, the default value is
+ * DEFAULT_PERCENTILE_ACCURACY.
+ */
+@ExpressionDescription(
+ usage =
+ """
+ _FUNC_(col, percentage [, accuracy]) - Returns the approximate percentile value of numeric
+ column `col` at the given percentage. The value of percentage must be between 0.0
+ and 1.0. The `accuracy` parameter (default: 10000) is a positive integer literal which
+ controls approximation accuracy at the cost of memory. Higher value of `accuracy` yields
+ better accuracy, `1.0/accuracy` is the relative error of the approximation.
+
+ _FUNC_(col, array(percentage1 [, percentage2]...) [, accuracy]) - Returns the approximate
+ percentile array of column `col` at the given percentage array. Each value of the
+ percentage array must be between 0.0 and 1.0. The `accuracy` parameter (default: 10000) is
+ a positive integer literal which controls approximation accuracy at the cost of memory.
+ Higher value of `accuracy` yields better accuracy, `1.0/accuracy` is the relative error of
+ the approximation.
+ """)
+case class ApproximatePercentile(
+ child: Expression,
+ percentageExpression: Expression,
+ accuracyExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int) extends TypedImperativeAggregate[PercentileDigest] {
+
+ def this(child: Expression, percentageExpression: Expression, accuracyExpression: Expression) = {
+ this(child, percentageExpression, accuracyExpression, 0, 0)
+ }
+
+ def this(child: Expression, percentageExpression: Expression) = {
+ this(child, percentageExpression, Literal(ApproximatePercentile.DEFAULT_PERCENTILE_ACCURACY))
+ }
+
+ // Mark as lazy so that accuracyExpression is not evaluated during tree transformation.
+ private lazy val accuracy: Int = accuracyExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(DoubleType, TypeCollection(DoubleType, ArrayType), IntegerType)
+ }
+
+ // Mark as lazy so that percentageExpression is not evaluated during tree transformation.
+ private lazy val (returnPercentileArray: Boolean, percentages: Array[Double]) = {
+ (percentageExpression.dataType, percentageExpression.eval()) match {
+ // Rule ImplicitTypeCasts can cast other numeric types to double
+ case (_, num: Double) => (false, Array(num))
+ case (ArrayType(baseType: NumericType, _), arrayData: ArrayData) =>
+ val numericArray = arrayData.toObjectArray(baseType)
+ (true, numericArray.map { x =>
+ baseType.numeric.toDouble(x.asInstanceOf[baseType.InternalType])
+ })
+ case other =>
+ throw new AnalysisException(s"Invalid data type ${other._1} for parameter percentage")
+ }
+ }
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (!percentageExpression.foldable || !accuracyExpression.foldable) {
+ TypeCheckFailure(s"The accuracy or percentage provided must be a constant literal")
+ } else if (accuracy <= 0) {
+ TypeCheckFailure(
+ s"The accuracy provided must be a positive integer literal (current value = $accuracy)")
+ } else if (percentages.exists(percentage => percentage < 0.0D || percentage > 1.0D)) {
+ TypeCheckFailure(
+ s"All percentage values must be between 0.0 and 1.0 " +
+ s"(current = ${percentages.mkString(", ")})")
+ } else {
+ TypeCheckSuccess
+ }
+ }
+
+ override def createAggregationBuffer(): PercentileDigest = {
+ val relativeError = 1.0D / accuracy
+ new PercentileDigest(relativeError)
+ }
+
+ override def update(buffer: PercentileDigest, inputRow: InternalRow): Unit = {
+ val value = child.eval(inputRow)
+ // Ignore empty rows, for example: percentile_approx(null)
+ if (value != null) {
+ buffer.add(value.asInstanceOf[Double])
+ }
+ }
+
+ override def merge(buffer: PercentileDigest, other: PercentileDigest): Unit = {
+ buffer.merge(other)
+ }
+
+ override def eval(buffer: PercentileDigest): Any = {
+ val result = buffer.getPercentiles(percentages)
+ if (result.length == 0) {
+ null
+ } else if (returnPercentileArray) {
+ new GenericArrayData(result)
+ } else {
+ result(0)
+ }
+ }
+
+ override def withNewMutableAggBufferOffset(newOffset: Int): ApproximatePercentile =
+ copy(mutableAggBufferOffset = newOffset)
+
+ override def withNewInputAggBufferOffset(newOffset: Int): ApproximatePercentile =
+ copy(inputAggBufferOffset = newOffset)
+
+ override def children: Seq[Expression] = Seq(child, percentageExpression, accuracyExpression)
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = {
+ if (returnPercentileArray) ArrayType(DoubleType) else DoubleType
+ }
+
+ override def prettyName: String = "percentile_approx"
+
+ override def serialize(obj: PercentileDigest): Array[Byte] = {
+ ApproximatePercentile.serializer.serialize(obj)
+ }
+
+ override def deserialize(bytes: Array[Byte]): PercentileDigest = {
+ ApproximatePercentile.serializer.deserialize(bytes)
+ }
+}
+
+object ApproximatePercentile {
+
+ // Default accuracy of Percentile approximation. Larger value means better accuracy.
+ // The default relative error can be deduced by defaultError = 1.0 / DEFAULT_PERCENTILE_ACCURACY
+ val DEFAULT_PERCENTILE_ACCURACY: Int = 10000
+
+ /**
+ * PercentileDigest is a probabilistic data structure used for approximating percentiles
+ * with limited memory. PercentileDigest is backed by [[QuantileSummaries]].
+ *
+ * @param summaries underlying probabilistic data structure [[QuantileSummaries]].
+ * @param isCompressed An internal flag from class [[QuantileSummaries]] to indicate whether the
+ * underlying quantileSummaries is compressed.
+ */
+ class PercentileDigest(
+ private var summaries: QuantileSummaries,
+ private var isCompressed: Boolean) {
--- End diff --
1. headSampled is private.
2. headSample.isEmpty not necessary means it is compressed. (Some bug in QuantileSummaries)
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: Implements percentile_approx aggregation function which ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14868
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/64646/
Test FAILed.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: Implements percentile_approx aggregation function which ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14868
**[Test build #64646 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64646/consoleFull)** for PR 14868 at commit [`96bd5a4`](https://github.com/apache/spark/commit/96bd5a4062d81995e438d0e1a301b42b7360994c).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14868: Implements percentile_approx aggregation function...
Posted by clockfly <gi...@git.apache.org>.
Github user clockfly commented on a diff in the pull request:
https://github.com/apache/spark/pull/14868#discussion_r76690870
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/QuantileSummaries.scala ---
@@ -41,7 +41,7 @@ import org.apache.spark.sql.catalyst.util.QuantileSummaries.Stats
* @param count the count of all the elements *inserted in the sampled buffer*
* (excluding the head buffer)
*/
-class QuantileSummaries(
+case class QuantileSummaries(
--- End diff --
Make it as a case class, so that we can use the expression encoder to serialize it.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: Implements percentile_approx aggregation function which ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14868
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/64623/
Test FAILed.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: Implements percentile_approx aggregation function which ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14868
**[Test build #64642 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64642/consoleFull)** for PR 14868 at commit [`2ab4a11`](https://github.com/apache/spark/commit/2ab4a116487a726f4ccd98a6c4fa30b0dd3bae8a).
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: Implements percentile_approx aggregation function which ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14868
**[Test build #64602 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64602/consoleFull)** for PR 14868 at commit [`6be2ebe`](https://github.com/apache/spark/commit/6be2ebe98b667e57511b9623062d54f43b04e617).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds the following public classes _(experimental)_:
* `case class ApproximatePercentile(`
* ` class PercentileDigest(`
* ` class PercentileDigestSerializer `
* ` case class QuantileSummariesData(`
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: Implements percentile_approx aggregation function which ...
Posted by clockfly <gi...@git.apache.org>.
Github user clockfly commented on the issue:
https://github.com/apache/spark/pull/14868
retest this please
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: Implements percentile_approx aggregation function which ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14868
**[Test build #64623 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64623/consoleFull)** for PR 14868 at commit [`9d7fb85`](https://github.com/apache/spark/commit/9d7fb85f6f2741f050344b06f2efac3254da77bf).
* This patch **fails to build**.
* This patch merges cleanly.
* This patch adds no public classes.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: Implements percentile_approx aggregation function which ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14868
**[Test build #64621 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64621/consoleFull)** for PR 14868 at commit [`dc09d8c`](https://github.com/apache/spark/commit/dc09d8c2f6191c98fefe4b9dfe7bdd817907560c).
* This patch **fails to build**.
* This patch merges cleanly.
* This patch adds no public classes.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14868: Implements percentile_approx aggregation function...
Posted by clockfly <gi...@git.apache.org>.
Github user clockfly commented on a diff in the pull request:
https://github.com/apache/spark/pull/14868#discussion_r76695365
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/ApproximatePercentileQuerySuite.scala ---
@@ -0,0 +1,226 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql
+
+import org.apache.spark.sql.catalyst.expressions.aggregate.ApproximatePercentile.DEFAULT_PERCENTILE_ACCURACY
+import org.apache.spark.sql.catalyst.expressions.aggregate.ApproximatePercentile.PercentileDigest
+import org.apache.spark.sql.test.SharedSQLContext
+
+class ApproximatePercentileQuerySuite extends QueryTest with SharedSQLContext {
+ import testImplicits._
+
+ private val table = "percentile_test"
+
+ test("percentile_approx, single percentile value") {
+ withTempView(table) {
+ (1 to 1000).toDF("col").createOrReplaceTempView(table)
+ checkAnswer(
+ spark.sql(
+ s"""
+ |SELECT
+ | percentile_approx(col, 0.25),
+ | percentile_approx(col, 0.5),
+ | percentile_approx(col, 0.75d),
+ | percentile_approx(col, 0.0),
+ | percentile_approx(col, 1.0),
+ | percentile_approx(col, 0),
+ | percentile_approx(col, 1)
+ |FROM $table
+ """.stripMargin),
+ Row(250D, 500D, 750D, 1D, 1000D, 1D, 1000D)
+ )
+ }
+ }
+
+ test("percentile_approx, array of percentile value") {
+ withTempView(table) {
+ (1 to 1000).toDF("col").createOrReplaceTempView(table)
+ checkAnswer(
+ spark.sql(
+ s"""SELECT
+ | percentile_approx(col, array(0.25, 0.5, 0.75D)),
+ | count(col),
+ | percentile_approx(col, array(0.0, 1.0)),
+ | sum(col)
+ |FROM $table
+ """.stripMargin),
+ Row(Seq(250D, 500D, 750D), 1000, Seq(1D, 1000D), 500500)
+ )
+ }
+ }
+
+ test("percentile_approx, with different accuracies") {
+
+ withTempView(table) {
+ (1 to 1000).toDF("col").createOrReplaceTempView(table)
+
+ // With different accuracies
+ val expectedPercentile = 250D
+ val accuracies = Array(1, 10, 100, 1000, 10000)
+ val errors = accuracies.map { accuracy =>
+ val df = spark.sql(s"SELECT percentile_approx(col, 0.25, $accuracy) FROM $table")
+ val approximatePercentile = df.collect().head.getDouble(0)
+ val error = Math.abs(approximatePercentile - expectedPercentile)
+ error
+ }
+
+ // The larger accuracy value we use, the smaller error we get
+ assert(errors.sorted.sameElements(errors.reverse))
+ }
+ }
+
+ test("percentile_approx, supports constant folding for parameter accuracy and percentages") {
+ withTempView(table) {
+ (1 to 1000).toDF("col").createOrReplaceTempView(table)
+ checkAnswer(
+ spark.sql(s"SELECT percentile_approx(col, array(0.25 + 0.25D), 200 + 800D) FROM $table"),
+ Row(Seq(500D))
+ )
+ }
+ }
+
+ test("percentile_approx(), aggregation on empty input table, no group by") {
+ withTempView(table) {
+ Seq.empty[Int].toDF("col").createOrReplaceTempView(table)
+ checkAnswer(
+ spark.sql(s"SELECT sum(col), percentile_approx(col, 0.5) FROM $table"),
+ Row(null, null)
+ )
+ }
+ }
+
+ test("percentile_approx(), aggregation on empty input table, with group by") {
+ withTempView(table) {
+ Seq.empty[Int].toDF("col").createOrReplaceTempView(table)
+ checkAnswer(
+ spark.sql(s"SELECT sum(col), percentile_approx(col, 0.5) FROM $table GROUP BY col"),
+ Seq.empty[Row]
+ )
+ }
+ }
+
+ test("percentile_approx(null), aggregation with group by") {
+ withTempView(table) {
+ (1 to 1000).map(x => (x % 3, x)).toDF("key", "value").createOrReplaceTempView(table)
+ checkAnswer(
+ spark.sql(
+ s"""SELECT
+ | key,
+ | percentile_approx(null, 0.5)
+ |FROM $table
+ |GROUP BY key
+ """.stripMargin),
+ Seq(
+ Row(0, null),
+ Row(1, null),
+ Row(2, null))
+ )
+ }
+ }
+
+ test("percentile_approx(null), aggregation without group by") {
+ withTempView(table) {
+ (1 to 1000).map(x => (x % 3, x)).toDF("key", "value").createOrReplaceTempView(table)
+ checkAnswer(
+ spark.sql(
+ s"""SELECT
+ | percentile_approx(null, 0.5),
+ | sum(null),
+ | percentile_approx(null, 0.5)
+ |FROM $table
+ """.stripMargin),
+ Row(null, null, null)
+ )
+ }
+ }
+
+ test("percentile_approx(col, ...), input rows contains null, with out group by") {
+ withTempView(table) {
+ (1 to 1000).map(new Integer(_)).flatMap(Seq(null: Integer, _)).toDF("col")
+ .createOrReplaceTempView(table)
+ checkAnswer(
+ spark.sql(
+ s"""SELECT
+ | percentile_approx(col, 0.5),
+ | sum(null),
+ | percentile_approx(col, 0.5)
+ |FROM $table
+ """.stripMargin),
+ Row(500D, null, 500D))
+ }
+ }
+
+ test("percentile_approx(col, ...), input rows contains null, with group by") {
+ withTempView(table) {
+ val rand = new java.util.Random()
+ (1 to 1000)
+ .map(new Integer(_))
+ .map(v => (new Integer(v % 2), v))
+ // Add some nulls
+ .flatMap(Seq(_, (null: Integer, null: Integer)))
+ .toDF("key", "value").createOrReplaceTempView(table)
+ checkAnswer(
+ spark.sql(
+ s"""SELECT
+ | percentile_approx(value, 0.5),
+ | sum(value),
+ | percentile_approx(value, 0.5)
+ |FROM $table
+ |GROUP BY key
+ """.stripMargin),
+ Seq(
+ Row(499.0D, 250000, 499.0D),
+ Row(500.0D, 250500, 500.0D),
+ Row(null, null, null))
+ )
+ }
+ }
+
+ test("percentile_approx(col, ...) works in window function") {
--- End diff --
TODO: The implementation of QuantileSummaries is not very clear when handling boundary (for example, add one record, and do immediate compression, and make a percentile query). We need to do a double check to make QuantileSummaries is correctly implemented.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: [SPARK-16283][SQL][WIP] Implements percentile_approx agg...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14868
Merged build finished. Test PASSed.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: Implements percentile_approx aggregation function which ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14868
**[Test build #64600 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64600/consoleFull)** for PR 14868 at commit [`88232ed`](https://github.com/apache/spark/commit/88232ed07780be3243c18ba1d3e7943cda628dcc).
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14868: Implements percentile_approx aggregation function...
Posted by liancheng <gi...@git.apache.org>.
Github user liancheng commented on a diff in the pull request:
https://github.com/apache/spark/pull/14868#discussion_r76761387
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/ApproximatePercentile.scala ---
@@ -0,0 +1,304 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions.aggregate
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.{InternalRow, ScalaReflectionLock}
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.{TypeCheckFailure, TypeCheckSuccess}
+import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.ApproximatePercentile.{PercentileDigest, PercentileDigestSerializer}
+import org.apache.spark.sql.catalyst.util.{ArrayData, GenericArrayData}
+import org.apache.spark.sql.catalyst.util.QuantileSummaries
+import org.apache.spark.sql.catalyst.util.QuantileSummaries.{defaultCompressThreshold, Stats}
+import org.apache.spark.sql.types._
+
+/**
+ * The ApproximatePercentile function returns the approximate percentile(s) of a column at the given
+ * percentage(s). A percentile is a watermark value below which a given percentage of the column
+ * values fall. For example, the percentile of column `col` at percentage 50% is the median of
+ * column `col`.
+ *
+ * This function supports partial aggregation.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param percentageExpression Expression that represents a single percentage value or
+ * a array of percentage values. Each percentage value must be between
--- End diff --
Nit: "a array" => "an array"
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: Implements percentile_approx aggregation function which ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14868
Merged build finished. Test FAILed.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: Implements percentile_approx aggregation function which ...
Posted by clockfly <gi...@git.apache.org>.
Github user clockfly commented on the issue:
https://github.com/apache/spark/pull/14868
The test failure is caused by https://issues.apache.org/jira/browse/SPARK-17289
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: Implements percentile_approx aggregation function which ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14868
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/64600/
Test FAILed.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14868: Implements percentile_approx aggregation function...
Posted by liancheng <gi...@git.apache.org>.
Github user liancheng commented on a diff in the pull request:
https://github.com/apache/spark/pull/14868#discussion_r76803012
--- Diff: sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/aggregate/ApproximatePercentileSuite.scala ---
@@ -0,0 +1,318 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions.aggregate
+
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.{SimpleAnalyzer, UnresolvedAttribute}
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.TypeCheckFailure
+import org.apache.spark.sql.catalyst.dsl.expressions._
+import org.apache.spark.sql.catalyst.dsl.plans._
+import org.apache.spark.sql.catalyst.expressions.{Alias, AttributeReference, BoundReference, Cast, CreateArray, DecimalLiteral, GenericMutableRow, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.ApproximatePercentile.{PercentileDigest, PercentileDigestSerializer}
+import org.apache.spark.sql.catalyst.plans.logical.LocalRelation
+import org.apache.spark.sql.catalyst.util.ArrayData
+import org.apache.spark.sql.catalyst.util.QuantileSummaries
+import org.apache.spark.sql.catalyst.util.QuantileSummaries.Stats
+import org.apache.spark.sql.types.{ArrayType, DoubleType, IntegerType}
+import org.apache.spark.util.SizeEstimator
+
+
+class ApproximatePercentileSuite extends SparkFunSuite {
+
+ private val random = new java.util.Random()
+
+ private val data = (0 until 10000).map { _ =>
+ random.nextInt(10000)
+ }
+
+ test("serialize and de-serialize") {
+
--- End diff --
Remove this new line.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: Implements percentile_approx aggregation function which ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14868
Merged build finished. Test FAILed.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: [SPARK-16283][SQL][WIP] Implements percentile_approx agg...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14868
Merged build finished. Test PASSed.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14868: [SPARK-16283][SQL] Implements percentile_approx a...
Posted by clockfly <gi...@git.apache.org>.
Github user clockfly commented on a diff in the pull request:
https://github.com/apache/spark/pull/14868#discussion_r77127003
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/ApproximatePercentile.scala ---
@@ -0,0 +1,321 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions.aggregate
+
+import java.nio.ByteBuffer
+
+import com.google.common.primitives.{Doubles, Ints, Longs}
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.{InternalRow}
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.{TypeCheckFailure, TypeCheckSuccess}
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.ApproximatePercentile.{PercentileDigest}
+import org.apache.spark.sql.catalyst.util.{ArrayData, GenericArrayData}
+import org.apache.spark.sql.catalyst.util.QuantileSummaries
+import org.apache.spark.sql.catalyst.util.QuantileSummaries.{defaultCompressThreshold, Stats}
+import org.apache.spark.sql.types._
+
+/**
+ * The ApproximatePercentile function returns the approximate percentile(s) of a column at the given
+ * percentage(s). A percentile is a watermark value below which a given percentage of the column
+ * values fall. For example, the percentile of column `col` at percentage 50% is the median of
+ * column `col`.
+ *
+ * This function supports partial aggregation.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param percentageExpression Expression that represents a single percentage value or
+ * an array of percentage values. Each percentage value must be between
+ * 0.0 and 1.0.
+ * @param accuracyExpression Integer literal expression of approximation accuracy. Higher value
+ * yields better accuracy, the default value is
+ * DEFAULT_PERCENTILE_ACCURACY.
+ */
+@ExpressionDescription(
+ usage =
+ """
+ _FUNC_(col, percentage [, accuracy]) - Returns the approximate percentile value of numeric
+ column `col` at the given percentage. The value of percentage must be between 0.0
+ and 1.0. The `accuracy` parameter (default: 10000) is a positive integer literal which
+ controls approximation accuracy at the cost of memory. Higher value of `accuracy` yields
+ better accuracy, `1.0/accuracy` is the relative error of the approximation.
+
+ _FUNC_(col, array(percentage1 [, percentage2]...) [, accuracy]) - Returns the approximate
+ percentile array of column `col` at the given percentage array. Each value of the
+ percentage array must be between 0.0 and 1.0. The `accuracy` parameter (default: 10000) is
+ a positive integer literal which controls approximation accuracy at the cost of memory.
+ Higher value of `accuracy` yields better accuracy, `1.0/accuracy` is the relative error of
+ the approximation.
+ """)
+case class ApproximatePercentile(
+ child: Expression,
+ percentageExpression: Expression,
+ accuracyExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int) extends TypedImperativeAggregate[PercentileDigest] {
+
+ def this(child: Expression, percentageExpression: Expression, accuracyExpression: Expression) = {
+ this(child, percentageExpression, accuracyExpression, 0, 0)
+ }
+
+ def this(child: Expression, percentageExpression: Expression) = {
+ this(child, percentageExpression, Literal(ApproximatePercentile.DEFAULT_PERCENTILE_ACCURACY))
+ }
+
+ // Mark as lazy so that accuracyExpression is not evaluated during tree transformation.
+ private lazy val accuracy: Int = accuracyExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(DoubleType, TypeCollection(DoubleType, ArrayType), IntegerType)
+ }
+
+ // Mark as lazy so that percentageExpression is not evaluated during tree transformation.
+ private lazy val (returnPercentileArray: Boolean, percentages: Array[Double]) = {
+ (percentageExpression.dataType, percentageExpression.eval()) match {
+ // Rule ImplicitTypeCasts can cast other numeric types to double
+ case (_, num: Double) => (false, Array(num))
+ case (ArrayType(baseType: NumericType, _), arrayData: ArrayData) =>
+ val numericArray = arrayData.toObjectArray(baseType)
+ (true, numericArray.map { x =>
+ baseType.numeric.toDouble(x.asInstanceOf[baseType.InternalType])
+ })
+ case other =>
+ throw new AnalysisException(s"Invalid data type ${other._1} for parameter percentage")
+ }
+ }
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (!percentageExpression.foldable || !accuracyExpression.foldable) {
+ TypeCheckFailure(s"The accuracy or percentage provided must be a constant literal")
+ } else if (accuracy <= 0) {
+ TypeCheckFailure(
+ s"The accuracy provided must be a positive integer literal (current value = $accuracy)")
+ } else if (percentages.exists(percentage => percentage < 0.0D || percentage > 1.0D)) {
+ TypeCheckFailure(
+ s"All percentage values must be between 0.0 and 1.0 " +
+ s"(current = ${percentages.mkString(", ")})")
+ } else {
+ TypeCheckSuccess
+ }
+ }
+
+ override def createAggregationBuffer(): PercentileDigest = {
+ val relativeError = 1.0D / accuracy
+ new PercentileDigest(relativeError)
+ }
+
+ override def update(buffer: PercentileDigest, inputRow: InternalRow): Unit = {
+ val value = child.eval(inputRow)
+ // Ignore empty rows, for example: percentile_approx(null)
+ if (value != null) {
+ buffer.add(value.asInstanceOf[Double])
+ }
+ }
+
+ override def merge(buffer: PercentileDigest, other: PercentileDigest): Unit = {
+ buffer.merge(other)
+ }
+
+ override def eval(buffer: PercentileDigest): Any = {
+ val result = buffer.getPercentiles(percentages)
+ if (result.length == 0) {
+ null
+ } else if (returnPercentileArray) {
+ new GenericArrayData(result)
+ } else {
+ result(0)
+ }
+ }
+
+ override def withNewMutableAggBufferOffset(newOffset: Int): ApproximatePercentile =
+ copy(mutableAggBufferOffset = newOffset)
+
+ override def withNewInputAggBufferOffset(newOffset: Int): ApproximatePercentile =
+ copy(inputAggBufferOffset = newOffset)
+
+ override def children: Seq[Expression] = Seq(child, percentageExpression, accuracyExpression)
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = {
+ if (returnPercentileArray) ArrayType(DoubleType) else DoubleType
+ }
+
+ override def prettyName: String = "percentile_approx"
+
+ override def serialize(obj: PercentileDigest): Array[Byte] = {
+ ApproximatePercentile.serializer.serialize(obj)
+ }
+
+ override def deserialize(bytes: Array[Byte]): PercentileDigest = {
+ ApproximatePercentile.serializer.deserialize(bytes)
+ }
+}
+
+object ApproximatePercentile {
+
+ // Default accuracy of Percentile approximation. Larger value means better accuracy.
+ // The default relative error can be deduced by defaultError = 1.0 / DEFAULT_PERCENTILE_ACCURACY
+ val DEFAULT_PERCENTILE_ACCURACY: Int = 10000
+
+ /**
+ * PercentileDigest is a probabilistic data structure used for approximating percentiles
+ * with limited memory. PercentileDigest is backed by [[QuantileSummaries]].
+ *
+ * @param summaries underlying probabilistic data structure [[QuantileSummaries]].
+ * @param isCompressed An internal flag from class [[QuantileSummaries]] to indicate whether the
+ * underlying quantileSummaries is compressed.
+ */
+ class PercentileDigest(
+ private var summaries: QuantileSummaries,
+ private var isCompressed: Boolean) {
+
+ // Trigger compression if the QuantileSummaries's buffer length exceeds
+ // compressThresHoldBufferLength. The buffer length can be get by
+ // quantileSummaries.sampled.length
+ private[this] final val compressThresHoldBufferLength: Int = {
+ // Max buffer length after compression.
+ val maxBufferLengthAfterCompression: Int = (1 / summaries.relativeError).toInt * 2
+ // A safe upper bound for buffer length before compression
+ maxBufferLengthAfterCompression * 2
+ }
+
+ def this(relativeError: Double) = {
+ this(new QuantileSummaries(defaultCompressThreshold, relativeError), isCompressed = true)
+ }
+
+ /** Returns compressed object of [[QuantileSummaries]] */
+ def quantileSummaries: QuantileSummaries = {
+ if (!isCompressed) compress()
+ summaries
+ }
+
+ /** Insert an observation value into the PercentileDigest data structure. */
+ def add(value: Double): Unit = {
+ summaries = summaries.insert(value)
+ // The result of QuantileSummaries.insert is un-compressed
+ isCompressed = false
+
+ // Currently, QuantileSummaries ignores the construction parameter compressThresHold,
+ // which may cause QuantileSummaries to occupy unbounded memory. We have to hack around here
+ // to make sure QuantileSummaries doesn't occupy infinite memory.
+ // TODO: Figure out why QuantileSummaries ignores construction parameter compressThresHold
+ if (summaries.sampled.length >= compressThresHoldBufferLength) compress()
+ }
+
+ /** In-place merges in another PercentileDigest. */
+ def merge(other: PercentileDigest): Unit = {
+ if (!isCompressed) compress()
+ summaries = summaries.merge(other.quantileSummaries)
--- End diff --
Yes. other.quantileSummaries will get a compression version of QuantileSummaries.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: [SPARK-16283][SQL][WIP] Implements percentile_approx agg...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14868
**[Test build #64696 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64696/consoleFull)** for PR 14868 at commit [`3f08c02`](https://github.com/apache/spark/commit/3f08c027add03c59251583420c76582a085b3573).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: Implements percentile_approx aggregation function which ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14868
Merged build finished. Test FAILed.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: Implements percentile_approx aggregation function which ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14868
**[Test build #64639 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64639/consoleFull)** for PR 14868 at commit [`9d7fb85`](https://github.com/apache/spark/commit/9d7fb85f6f2741f050344b06f2efac3254da77bf).
* This patch **fails to build**.
* This patch merges cleanly.
* This patch adds no public classes.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: Implements percentile_approx aggregation function which ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14868
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/64639/
Test FAILed.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: [SPARK-16283][SQL][WIP] Implements percentile_approx agg...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14868
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/64695/
Test PASSed.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: [SPARK-16283][SQL][WIP] Implements percentile_approx agg...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14868
**[Test build #64695 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64695/consoleFull)** for PR 14868 at commit [`bc70a00`](https://github.com/apache/spark/commit/bc70a0023bb24175c06c03cb7acad7f9a6d34e36).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: [SPARK-16283][SQL] Implements percentile_approx aggregat...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on the issue:
https://github.com/apache/spark/pull/14868
Although the function returns the approximate percentile, the result is still deterministic. Is my understanding right?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: Implements percentile_approx aggregation function which ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14868
**[Test build #64602 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64602/consoleFull)** for PR 14868 at commit [`6be2ebe`](https://github.com/apache/spark/commit/6be2ebe98b667e57511b9623062d54f43b04e617).
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: Implements percentile_approx aggregation function which ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14868
Merged build finished. Test FAILed.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14868: Implements percentile_approx aggregation function...
Posted by liancheng <gi...@git.apache.org>.
Github user liancheng commented on a diff in the pull request:
https://github.com/apache/spark/pull/14868#discussion_r76761771
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/ApproximatePercentile.scala ---
@@ -0,0 +1,304 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions.aggregate
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.{InternalRow, ScalaReflectionLock}
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.{TypeCheckFailure, TypeCheckSuccess}
+import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.ApproximatePercentile.{PercentileDigest, PercentileDigestSerializer}
+import org.apache.spark.sql.catalyst.util.{ArrayData, GenericArrayData}
+import org.apache.spark.sql.catalyst.util.QuantileSummaries
+import org.apache.spark.sql.catalyst.util.QuantileSummaries.{defaultCompressThreshold, Stats}
+import org.apache.spark.sql.types._
+
+/**
+ * The ApproximatePercentile function returns the approximate percentile(s) of a column at the given
+ * percentage(s). A percentile is a watermark value below which a given percentage of the column
+ * values fall. For example, the percentile of column `col` at percentage 50% is the median of
+ * column `col`.
+ *
+ * This function supports partial aggregation.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param percentageExpression Expression that represents a single percentage value or
+ * a array of percentage values. Each percentage value must be between
+ * 0.0 and 1.0.
+ * @param accuracyExpression Integer literal expression of approximation accuracy. Higher value
+ * yield better accuracy, the default value is
+ * DEFAULT_PERCENTILE_ACCURACY.
+ */
+@ExpressionDescription(
+ usage =
+ """
+ _FUNC_(col, percentage [, accuracy]) - Returns the approximate percentile value of numeric
+ column `col` at the given percentage. The value of percentage must be between 0.0
+ and 1.0. The `accuracy` parameter (default: 10000) is a positive integer literal which
+ controls approximation accuracy at the cost of memory. Higher value of `accuracy` yield
+ better accuracy, `1.0/accuracy` is the relative error of the approximation.
+
+ _FUNC_(col, array(percentage1 [, percentage2]...) [, accuracy]) - Returns the approximate
+ percentile array of column `col` at the given percentage array. Each value of the
+ percentage array must be between 0.0 and 1.0. The `accuracy` parameter (default: 10000) is
+ a positive integer literal which controls approximation accuracy at the cost of memory.
+ Higher value of `accuracy` yield better accuracy, `1.0/accuracy` is the relative error of
--- End diff --
Same as above.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: Implements percentile_approx aggregation function which ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14868
**[Test build #64643 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64643/consoleFull)** for PR 14868 at commit [`3dda1ae`](https://github.com/apache/spark/commit/3dda1ae90cea0eab8dd963d4af5708bff8846d65).
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: Implements percentile_approx aggregation function which ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14868
Merged build finished. Test FAILed.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: Implements percentile_approx aggregation function which ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14868
**[Test build #64676 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64676/consoleFull)** for PR 14868 at commit [`1965b1a`](https://github.com/apache/spark/commit/1965b1a918633f0d9b94b0a245605592cddbeb9b).
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: [SPARK-16283][SQL][WIP] Implements percentile_approx agg...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14868
**[Test build #64696 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64696/consoleFull)** for PR 14868 at commit [`3f08c02`](https://github.com/apache/spark/commit/3f08c027add03c59251583420c76582a085b3573).
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: [SPARK-16283][SQL][WIP] Implements percentile_approx agg...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14868
**[Test build #64676 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64676/consoleFull)** for PR 14868 at commit [`1965b1a`](https://github.com/apache/spark/commit/1965b1a918633f0d9b94b0a245605592cddbeb9b).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14868: [SPARK-16283][SQL][WIP] Implements percentile_app...
Posted by clockfly <gi...@git.apache.org>.
Github user clockfly commented on a diff in the pull request:
https://github.com/apache/spark/pull/14868#discussion_r76918794
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/ApproximatePercentile.scala ---
@@ -0,0 +1,321 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions.aggregate
+
+import java.nio.ByteBuffer
+
+import com.google.common.primitives.{Doubles, Ints, Longs}
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.{InternalRow}
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.{TypeCheckFailure, TypeCheckSuccess}
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.ApproximatePercentile.{PercentileDigest}
+import org.apache.spark.sql.catalyst.util.{ArrayData, GenericArrayData}
+import org.apache.spark.sql.catalyst.util.QuantileSummaries
+import org.apache.spark.sql.catalyst.util.QuantileSummaries.{defaultCompressThreshold, Stats}
+import org.apache.spark.sql.types._
+
+/**
+ * The ApproximatePercentile function returns the approximate percentile(s) of a column at the given
+ * percentage(s). A percentile is a watermark value below which a given percentage of the column
+ * values fall. For example, the percentile of column `col` at percentage 50% is the median of
+ * column `col`.
+ *
+ * This function supports partial aggregation.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param percentageExpression Expression that represents a single percentage value or
+ * an array of percentage values. Each percentage value must be between
+ * 0.0 and 1.0.
+ * @param accuracyExpression Integer literal expression of approximation accuracy. Higher value
+ * yields better accuracy, the default value is
+ * DEFAULT_PERCENTILE_ACCURACY.
+ */
+@ExpressionDescription(
+ usage =
+ """
+ _FUNC_(col, percentage [, accuracy]) - Returns the approximate percentile value of numeric
+ column `col` at the given percentage. The value of percentage must be between 0.0
+ and 1.0. The `accuracy` parameter (default: 10000) is a positive integer literal which
+ controls approximation accuracy at the cost of memory. Higher value of `accuracy` yields
+ better accuracy, `1.0/accuracy` is the relative error of the approximation.
+
+ _FUNC_(col, array(percentage1 [, percentage2]...) [, accuracy]) - Returns the approximate
+ percentile array of column `col` at the given percentage array. Each value of the
+ percentage array must be between 0.0 and 1.0. The `accuracy` parameter (default: 10000) is
+ a positive integer literal which controls approximation accuracy at the cost of memory.
+ Higher value of `accuracy` yields better accuracy, `1.0/accuracy` is the relative error of
+ the approximation.
+ """)
+case class ApproximatePercentile(
+ child: Expression,
+ percentageExpression: Expression,
+ accuracyExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int) extends TypedImperativeAggregate[PercentileDigest] {
+
+ def this(child: Expression, percentageExpression: Expression, accuracyExpression: Expression) = {
+ this(child, percentageExpression, accuracyExpression, 0, 0)
+ }
+
+ def this(child: Expression, percentageExpression: Expression) = {
+ this(child, percentageExpression, Literal(ApproximatePercentile.DEFAULT_PERCENTILE_ACCURACY))
+ }
+
+ // Mark as lazy so that accuracyExpression is not evaluated during tree transformation.
+ private lazy val accuracy: Int = accuracyExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(DoubleType, TypeCollection(DoubleType, ArrayType), IntegerType)
+ }
+
+ // Mark as lazy so that percentageExpression is not evaluated during tree transformation.
+ private lazy val (returnPercentileArray: Boolean, percentages: Array[Double]) = {
+ (percentageExpression.dataType, percentageExpression.eval()) match {
+ // Rule ImplicitTypeCasts can cast other numeric types to double
+ case (_, num: Double) => (false, Array(num))
+ case (ArrayType(baseType: NumericType, _), arrayData: ArrayData) =>
+ val numericArray = arrayData.toObjectArray(baseType)
+ (true, numericArray.map { x =>
+ baseType.numeric.toDouble(x.asInstanceOf[baseType.InternalType])
+ })
+ case other =>
+ throw new AnalysisException(s"Invalid data type ${other._1} for parameter percentage")
+ }
+ }
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (!percentageExpression.foldable || !accuracyExpression.foldable) {
+ TypeCheckFailure(s"The accuracy or percentage provided must be a constant literal")
+ } else if (accuracy <= 0) {
+ TypeCheckFailure(
+ s"The accuracy provided must be a positive integer literal (current value = $accuracy)")
+ } else if (percentages.exists(percentage => percentage < 0.0D || percentage > 1.0D)) {
+ TypeCheckFailure(
+ s"All percentage values must be between 0.0 and 1.0 " +
+ s"(current = ${percentages.mkString(", ")})")
+ } else {
+ TypeCheckSuccess
+ }
+ }
+
+ override def createAggregationBuffer(): PercentileDigest = {
+ val relativeError = 1.0D / accuracy
+ new PercentileDigest(relativeError)
+ }
+
+ override def update(buffer: PercentileDigest, inputRow: InternalRow): Unit = {
+ val value = child.eval(inputRow)
+ // Ignore empty rows, for example: percentile_approx(null)
+ if (value != null) {
+ buffer.add(value.asInstanceOf[Double])
+ }
+ }
+
+ override def merge(buffer: PercentileDigest, other: PercentileDigest): Unit = {
+ buffer.merge(other)
+ }
+
+ override def eval(buffer: PercentileDigest): Any = {
+ val result = buffer.getPercentiles(percentages)
+ if (result.length == 0) {
+ null
+ } else if (returnPercentileArray) {
+ new GenericArrayData(result)
+ } else {
+ result(0)
+ }
+ }
+
+ override def withNewMutableAggBufferOffset(newOffset: Int): ApproximatePercentile =
+ copy(mutableAggBufferOffset = newOffset)
+
+ override def withNewInputAggBufferOffset(newOffset: Int): ApproximatePercentile =
+ copy(inputAggBufferOffset = newOffset)
+
+ override def children: Seq[Expression] = Seq(child, percentageExpression, accuracyExpression)
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = {
+ if (returnPercentileArray) ArrayType(DoubleType) else DoubleType
+ }
+
+ override def prettyName: String = "percentile_approx"
+
+ override def serialize(obj: PercentileDigest): Array[Byte] = {
+ ApproximatePercentile.serializer.serialize(obj)
+ }
+
+ override def deserialize(bytes: Array[Byte]): PercentileDigest = {
+ ApproximatePercentile.serializer.deserialize(bytes)
+ }
+}
+
+object ApproximatePercentile {
+
+ // Default accuracy of Percentile approximation. Larger value means better accuracy.
+ // The default relative error can be deduced by defaultError = 1.0 / DEFAULT_PERCENTILE_ACCURACY
+ val DEFAULT_PERCENTILE_ACCURACY: Int = 10000
+
+ /**
+ * PercentileDigest is a probabilistic data structure used for approximating percentiles
+ * with limited memory. PercentileDigest is backed by [[QuantileSummaries]].
+ *
+ * @param summaries underlying probabilistic data structure [[QuantileSummaries]].
+ * @param isCompressed An internal flag from class [[QuantileSummaries]] to indicate whether the
+ * underlying quantileSummaries is compressed.
+ */
+ class PercentileDigest(
+ private var summaries: QuantileSummaries,
+ private var isCompressed: Boolean) {
+
+ // Trigger compression if the QuantileSummaries's buffer length exceeds
+ // compressThresHoldBufferLength. The buffer length can be get by
+ // quantileSummaries.sampled.length
+ private[this] final val compressThresHoldBufferLength: Int = {
+ // Max buffer length after compression.
+ val maxBufferLengthAfterCompression: Int = (1 / summaries.relativeError).toInt * 2
+ // A safe upper bound for buffer length before compression
+ maxBufferLengthAfterCompression * 2
+ }
+
+ def this(relativeError: Double) = {
+ this(new QuantileSummaries(defaultCompressThreshold, relativeError), isCompressed = true)
+ }
+
+ /** Returns compressed object of [[QuantileSummaries]] */
+ def quantileSummaries: QuantileSummaries = {
+ if (!isCompressed) compress()
+ summaries
+ }
+
+ /** Insert an observation value into the PercentileDigest data structure. */
+ def add(value: Double): Unit = {
+ summaries = summaries.insert(value)
+ // The result of QuantileSummaries.insert is un-compressed
+ isCompressed = false
+
+ // Currently, QuantileSummaries ignores the construction parameter compressThresHold,
+ // which may cause QuantileSummaries to occupy unbounded memory. We have to hack around here
+ // to make sure QuantileSummaries doesn't occupy infinite memory.
+ // TODO: Figure out why QuantileSummaries ignores construction parameter compressThresHold
+ if (summaries.sampled.length >= compressThresHoldBufferLength) compress()
+ }
+
+ /** In-place merges in another PercentileDigest. */
+ def merge(other: PercentileDigest): Unit = {
+ if (!isCompressed) compress()
+ summaries = summaries.merge(other.quantileSummaries)
+ }
+
+ /**
+ * Returns the approximate percentiles of all observation values at the given percentages.
+ * A percentile is a watermark value below which a given percentage of observation values fall.
+ * For example, the following code returns the 25th, median, and 75th percentiles of
+ * all observation values:
+ *
+ * {{{
+ * val Array(p25, median, p75) = percentileDigest.getPercentiles(Array(0.25, 0.5, 0.75))
+ * }}}
+ */
+ def getPercentiles(percentages: Array[Double]): Array[Double] = {
+ if (!isCompressed) compress()
+ if (summaries.count == 0 || percentages.length == 0) {
+ Array.empty[Double]
+ } else {
+ val result = new Array[Double](percentages.length)
+ var i = 0
+ while (i < percentages.length) {
+ result(i) = summaries.query(percentages(i))
+ i += 1
+ }
+ result
+ }
+ }
+
+ private final def compress(): Unit = {
+ summaries = summaries.compress()
+ isCompressed = true
+ }
+ }
+
+ /**
+ * Serializer for class [[PercentileDigest]]
+ *
+ * This class is thread safe.
+ */
+ class PercentileDigestSerializer {
+
+ private final def length(summaries: QuantileSummaries): Int = {
+ // summaries.compressThreshold, summary.relativeError, summary.count
+ Ints.BYTES + Doubles.BYTES + Longs.BYTES +
+ // length of summary.sampled
+ Ints.BYTES +
+ // summary.sampled, Array[Stat(value: Double, g: Int, delta: Int)]
+ summaries.sampled.length * (Doubles.BYTES + Ints.BYTES + Ints.BYTES)
+ }
+
+ final def serialize(obj: PercentileDigest): Array[Byte] = {
--- End diff --
I swap the serialization implementation from ExpressionEncoder because it is not safe to initialize ExpressionEncoder at the Executor side.
(scala reflection doesn't works well with ExecutorClassLoader)
stacktrace:
```
scala> spark.sql("select percentile_approx(a, 0.5) from t").show
16/08/31 09:17:58 WARN TaskSetManager: Stage 0 contains a task of very large size (327 KB). The maximum recommended task size is 100 KB.
16/08/31 09:17:59 ERROR ExecutorClassLoader: Failed to check existence of class <root>.package on REPL class server at spark://127.0.0.1:58512/classes
java.net.URISyntaxException: Illegal character in path at index 32: spark://127.0.0.1:58512/classes/<root>/package.class
at java.net.URI$Parser.fail(URI.java:2848)
at java.net.URI$Parser.checkChars(URI.java:3021)
at java.net.URI$Parser.parseHierarchical(URI.java:3105)
at java.net.URI$Parser.parse(URI.java:3053)
at java.net.URI.<init>(URI.java:588)
at org.apache.spark.rpc.netty.NettyRpcEnv.openChannel(NettyRpcEnv.scala:316)
at org.apache.spark.repl.ExecutorClassLoader.org$apache$spark$repl$ExecutorClassLoader$$getClassFileInputStreamFromSparkRPC(ExecutorClassLoader.scala:90)
at org.apache.spark.repl.ExecutorClassLoader$$anonfun$1.apply(ExecutorClassLoader.scala:57)
at org.apache.spark.repl.ExecutorClassLoader$$anonfun$1.apply(ExecutorClassLoader.scala:57)
at org.apache.spark.repl.ExecutorClassLoader.findClassLocally(ExecutorClassLoader.scala:162)
at org.apache.spark.repl.ExecutorClassLoader.findClass(ExecutorClassLoader.scala:80)
at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
at java.lang.Class.forName0(Native Method)
at java.lang.Class.forName(Class.java:348)
at scala.reflect.runtime.JavaMirrors$JavaMirror.javaClass(JavaMirrors.scala:555)
at scala.reflect.runtime.JavaMirrors$JavaMirror.tryJavaClass(JavaMirrors.scala:559)
at scala.reflect.runtime.SymbolLoaders$PackageScope$$anonfun$lookupEntry$1.apply(SymbolLoaders.scala:137)
at scala.reflect.runtime.SymbolLoaders$PackageScope$$anonfun$lookupEntry$1.apply(SymbolLoaders.scala:126)
at scala.reflect.runtime.Gil$class.gilSynchronized(Gil.scala:19)
at scala.reflect.runtime.JavaUniverse.gilSynchronized(JavaUniverse.scala:16)
at scala.reflect.runtime.SymbolLoaders$PackageScope.syncLockSynchronized(SymbolLoaders.scala:124)
at scala.reflect.runtime.SymbolLoaders$PackageScope.lookupEntry(SymbolLoaders.scala:126)
at scala.reflect.internal.Types$Type.findDecl(Types.scala:971)
at scala.reflect.internal.Types$Type.decl(Types.scala:566)
at scala.reflect.internal.SymbolTable.openPackageModule(SymbolTable.scala:335)
at scala.reflect.runtime.SymbolLoaders$LazyPackageType$$anonfun$complete$2.apply$mcV$sp(SymbolLoaders.scala:74)
at scala.reflect.runtime.SymbolLoaders$LazyPackageType$$anonfun$complete$2.apply(SymbolLoaders.scala:71)
at scala.reflect.runtime.SymbolLoaders$LazyPackageType$$anonfun$complete$2.apply(SymbolLoaders.scala:71)
at scala.reflect.internal.SymbolTable.slowButSafeEnteringPhaseNotLaterThan(SymbolTable.scala:263)
at scala.reflect.runtime.SymbolLoaders$LazyPackageType.complete(SymbolLoaders.scala:71)
at scala.reflect.internal.Symbols$Symbol.info(Symbols.scala:1514)
at scala.reflect.runtime.JavaMirrors$JavaMirror$$anon$1.scala$reflect$runtime$SynchronizedSymbols$SynchronizedSymbol$$super$info(JavaMirrors.scala:66)
at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$$anonfun$info$1.apply(SynchronizedSymbols.scala:127)
at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$$anonfun$info$1.apply(SynchronizedSymbols.scala:127)
at scala.reflect.runtime.Gil$class.gilSynchronized(Gil.scala:19)
at scala.reflect.runtime.JavaUniverse.gilSynchronized(JavaUniverse.scala:16)
at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$class.gilSynchronizedIfNotThreadsafe(SynchronizedSymbols.scala:123)
at scala.reflect.runtime.JavaMirrors$JavaMirror$$anon$1.gilSynchronizedIfNotThreadsafe(JavaMirrors.scala:66)
at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$class.info(SynchronizedSymbols.scala:127)
at scala.reflect.runtime.JavaMirrors$JavaMirror$$anon$1.info(JavaMirrors.scala:66)
at scala.reflect.internal.Mirrors$RootsBase.init(Mirrors.scala:256)
at scala.reflect.runtime.JavaMirrors$class.scala$reflect$runtime$JavaMirrors$$createMirror(JavaMirrors.scala:32)
at scala.reflect.runtime.JavaMirrors$$anonfun$runtimeMirror$1.apply(JavaMirrors.scala:49)
at scala.reflect.runtime.JavaMirrors$$anonfun$runtimeMirror$1.apply(JavaMirrors.scala:47)
at scala.reflect.runtime.Gil$class.gilSynchronized(Gil.scala:19)
at scala.reflect.runtime.JavaUniverse.gilSynchronized(JavaUniverse.scala:16)
at scala.reflect.runtime.JavaMirrors$class.runtimeMirror(JavaMirrors.scala:46)
at scala.reflect.runtime.JavaUniverse.runtimeMirror(JavaUniverse.scala:16)
at scala.reflect.runtime.JavaUniverse.runtimeMirror(JavaUniverse.scala:16)
at org.apache.spark.sql.catalyst.ScalaReflection$.mirror(ScalaReflection.scala:46)
at org.apache.spark.sql.catalyst.ScalaReflection$.mirror(ScalaReflection.scala:39)
at org.apache.spark.sql.catalyst.ScalaReflection$class.localTypeOf(ScalaReflection.scala:753)
at org.apache.spark.sql.catalyst.ScalaReflection$.localTypeOf(ScalaReflection.scala:39)
at org.apache.spark.sql.catalyst.ScalaReflection$.definedByConstructorParams(ScalaReflection.scala:712)
at org.apache.spark.sql.catalyst.encoders.ExpressionEncoder$.apply(ExpressionEncoder.scala:51)
at org.apache.spark.sql.catalyst.expressions.aggregate.ApproximatePercentile$PercentileDigestSerializer.<init>(ApproximatePercentile.scala:273)
at org.apache.spark.sql.catalyst.expressions.aggregate.ApproximatePercentile.serializer$lzycompute(ApproximatePercentile.scala:82)
at org.apache.spark.sql.catalyst.expressions.aggregate.ApproximatePercentile.serializer(ApproximatePercentile.scala:82)
at org.apache.spark.sql.catalyst.expressions.aggregate.ApproximatePercentile.serialize(ApproximatePercentile.scala:167)
at org.apache.spark.sql.catalyst.expressions.aggregate.ApproximatePercentile.serialize(ApproximatePercentile.scala:64)
at org.apache.spark.sql.catalyst.expressions.aggregate.TypedImperativeAggregate.serializeAggregateBufferInPlace(interfaces.scala:530)
at org.apache.spark.sql.execution.aggregate.AggregationIterator$$anonfun$generateResultProjection$2.apply(AggregationIterator.scala:250)
at org.apache.spark.sql.execution.aggregate.AggregationIterator$$anonfun$generateResultProjection$2.apply(AggregationIterator.scala:246)
at org.apache.spark.sql.execution.aggregate.SortBasedAggregationIterator.next(SortBasedAggregationIterator.scala:152)
at org.apache.spark.sql.execution.aggregate.SortBasedAggregationIterator.next(SortBasedAggregationIterator.scala:29)
at scala.collection.Iterator$$anon$11.next(Iterator.scala:409)
at org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.write(BypassMergeSortShuffleWriter.java:150)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:79)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:47)
at org.apache.spark.scheduler.Task.run(Task.scala:86)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:274)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
16/08/31 09:17:59 ERROR ExecutorClassLoader: Failed to check existence of class <root>.scala on REPL class server at spark://127.0.0.1:58512/classes
java.net.URISyntaxException: Illegal character in path at index 32: spark://127.0.0.1:58512/classes/<root>/scala.class
at java.net.URI$Parser.fail(URI.java:2848)
at java.net.URI$Parser.checkChars(URI.java:3021)
at java.net.URI$Parser.parseHierarchical(URI.java:3105)
at java.net.URI$Parser.parse(URI.java:3053)
at java.net.URI.<init>(URI.java:588)
at org.apache.spark.rpc.netty.NettyRpcEnv.openChannel(NettyRpcEnv.scala:316)
at org.apache.spark.repl.ExecutorClassLoader.org$apache$spark$repl$ExecutorClassLoader$$getClassFileInputStreamFromSparkRPC(ExecutorClassLoader.scala:90)
at org.apache.spark.repl.ExecutorClassLoader$$anonfun$1.apply(ExecutorClassLoader.scala:57)
at org.apache.spark.repl.ExecutorClassLoader$$anonfun$1.apply(ExecutorClassLoader.scala:57)
at org.apache.spark.repl.ExecutorClassLoader.findClassLocally(ExecutorClassLoader.scala:162)
at org.apache.spark.repl.ExecutorClassLoader.findClass(ExecutorClassLoader.scala:80)
at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
at java.lang.Class.forName0(Native Method)
at java.lang.Class.forName(Class.java:348)
at scala.reflect.runtime.JavaMirrors$JavaMirror.javaClass(JavaMirrors.scala:555)
at scala.reflect.runtime.JavaMirrors$JavaMirror.tryJavaClass(JavaMirrors.scala:559)
at scala.reflect.runtime.SymbolLoaders$PackageScope$$anonfun$lookupEntry$1.apply(SymbolLoaders.scala:137)
at scala.reflect.runtime.SymbolLoaders$PackageScope$$anonfun$lookupEntry$1.apply(SymbolLoaders.scala:126)
at scala.reflect.runtime.Gil$class.gilSynchronized(Gil.scala:19)
at scala.reflect.runtime.JavaUniverse.gilSynchronized(JavaUniverse.scala:16)
at scala.reflect.runtime.SymbolLoaders$PackageScope.syncLockSynchronized(SymbolLoaders.scala:124)
at scala.reflect.runtime.SymbolLoaders$PackageScope.lookupEntry(SymbolLoaders.scala:126)
at scala.reflect.internal.tpe.FindMembers$FindMemberBase.walkBaseClasses(FindMembers.scala:88)
at scala.reflect.internal.tpe.FindMembers$FindMemberBase.searchConcreteThenDeferred(FindMembers.scala:56)
at scala.reflect.internal.tpe.FindMembers$FindMemberBase.apply(FindMembers.scala:48)
at scala.reflect.internal.Types$Type.scala$reflect$internal$Types$Type$$findMemberInternal$1(Types.scala:1014)
at scala.reflect.internal.Types$Type.findMember(Types.scala:1016)
at scala.reflect.internal.Types$Type.memberBasedOnName(Types.scala:631)
at scala.reflect.internal.Types$Type.member(Types.scala:600)
at scala.reflect.internal.Mirrors$RootsBase.getModuleOrClass(Mirrors.scala:48)
at scala.reflect.internal.Mirrors$RootsBase.getModuleOrClass(Mirrors.scala:45)
at scala.reflect.internal.Mirrors$RootsBase.getModuleOrClass(Mirrors.scala:66)
at scala.reflect.internal.Mirrors$RootsBase.staticPackage(Mirrors.scala:204)
at scala.reflect.runtime.JavaMirrors$JavaMirror.staticPackage(JavaMirrors.scala:82)
at scala.reflect.internal.Mirrors$RootsBase.init(Mirrors.scala:263)
at scala.reflect.runtime.JavaMirrors$class.scala$reflect$runtime$JavaMirrors$$createMirror(JavaMirrors.scala:32)
at scala.reflect.runtime.JavaMirrors$$anonfun$runtimeMirror$1.apply(JavaMirrors.scala:49)
at scala.reflect.runtime.JavaMirrors$$anonfun$runtimeMirror$1.apply(JavaMirrors.scala:47)
at scala.reflect.runtime.Gil$class.gilSynchronized(Gil.scala:19)
at scala.reflect.runtime.JavaUniverse.gilSynchronized(JavaUniverse.scala:16)
at scala.reflect.runtime.JavaMirrors$class.runtimeMirror(JavaMirrors.scala:46)
at scala.reflect.runtime.JavaUniverse.runtimeMirror(JavaUniverse.scala:16)
at scala.reflect.runtime.JavaUniverse.runtimeMirror(JavaUniverse.scala:16)
at org.apache.spark.sql.catalyst.ScalaReflection$.mirror(ScalaReflection.scala:46)
at org.apache.spark.sql.catalyst.ScalaReflection$.mirror(ScalaReflection.scala:39)
at org.apache.spark.sql.catalyst.ScalaReflection$class.localTypeOf(ScalaReflection.scala:753)
at org.apache.spark.sql.catalyst.ScalaReflection$.localTypeOf(ScalaReflection.scala:39)
at org.apache.spark.sql.catalyst.ScalaReflection$.definedByConstructorParams(ScalaReflection.scala:712)
at org.apache.spark.sql.catalyst.encoders.ExpressionEncoder$.apply(ExpressionEncoder.scala:51)
at org.apache.spark.sql.catalyst.expressions.aggregate.ApproximatePercentile$PercentileDigestSerializer.<init>(ApproximatePercentile.scala:273)
at org.apache.spark.sql.catalyst.expressions.aggregate.ApproximatePercentile.serializer$lzycompute(ApproximatePercentile.scala:82)
at org.apache.spark.sql.catalyst.expressions.aggregate.ApproximatePercentile.serializer(ApproximatePercentile.scala:82)
at org.apache.spark.sql.catalyst.expressions.aggregate.ApproximatePercentile.serialize(ApproximatePercentile.scala:167)
at org.apache.spark.sql.catalyst.expressions.aggregate.ApproximatePercentile.serialize(ApproximatePercentile.scala:64)
at org.apache.spark.sql.catalyst.expressions.aggregate.TypedImperativeAggregate.serializeAggregateBufferInPlace(interfaces.scala:530)
at org.apache.spark.sql.execution.aggregate.AggregationIterator$$anonfun$generateResultProjection$2.apply(AggregationIterator.scala:250)
at org.apache.spark.sql.execution.aggregate.AggregationIterator$$anonfun$generateResultProjection$2.apply(AggregationIterator.scala:246)
at org.apache.spark.sql.execution.aggregate.SortBasedAggregationIterator.next(SortBasedAggregationIterator.scala:152)
at org.apache.spark.sql.execution.aggregate.SortBasedAggregationIterator.next(SortBasedAggregationIterator.scala:29)
at scala.collection.Iterator$$anon$11.next(Iterator.scala:409)
at org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.write(BypassMergeSortShuffleWriter.java:150)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:79)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:47)
at org.apache.spark.scheduler.Task.run(Task.scala:86)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:274)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
```
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: [SPARK-16283][SQL][WIP] Implements percentile_approx agg...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14868
**[Test build #64687 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64687/consoleFull)** for PR 14868 at commit [`421e211`](https://github.com/apache/spark/commit/421e2118d3883ff57b01704327f887eb6d6ea3e8).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14868: Implements percentile_approx aggregation function...
Posted by clockfly <gi...@git.apache.org>.
Github user clockfly commented on a diff in the pull request:
https://github.com/apache/spark/pull/14868#discussion_r76695042
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/ApproximatePercentile.scala ---
@@ -0,0 +1,302 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions.aggregate
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.{InternalRow, ScalaReflectionLock}
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.{TypeCheckFailure, TypeCheckSuccess}
+import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.ApproximatePercentile.{PercentileDigest, PercentileDigestSerializer}
+import org.apache.spark.sql.catalyst.util.{ArrayData, GenericArrayData}
+import org.apache.spark.sql.catalyst.util.QuantileSummaries
+import org.apache.spark.sql.catalyst.util.QuantileSummaries.{defaultCompressThreshold, Stats}
+import org.apache.spark.sql.types._
+
+/**
+ * The ApproximatePercentile function returns the approximate percentile(s) of a column at the given
+ * percentage(s). A percentile is a watermark value below which a given percentage of the column
+ * values fall. For example, the percentile of column `col` at percentage 50% is the median of
+ * column `col`.
+ *
+ * This function supports partial aggregation.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param percentageExpression Expression that represents a single percentage value or
+ * a array of percentage values. Each percentage value must be between
+ * 0.0 and 1.0.
+ * @param accuracyExpression Integer literal expression of approximation accuracy. Higher value
+ * yield better accuracy, the default value is
+ * DEFAULT_PERCENTILE_ACCURACY.
+ */
+@ExpressionDescription(
+ usage =
+ """
+ _FUNC_(col, percentage [, accuracy]) - Returns the approximate percentile value of numeric
+ column `col` at the given percentage. The value of percentage must be between 0.0
+ and 1.0. The `accuracy` parameter (default: 10000) is a positive integer literal which
+ controls approximation accuracy at the cost of memory. Higher value of `accuracy` yield
+ better accuracy, `1.0/accuracy` is the relative error of the approximation.
+
+ _FUNC_(col, array(percentage1 [, percentage2]...) [, accuracy]) - Returns the approximate
+ percentile array of column `col` at the given percentage array. Each value of the
+ percentage array must be between 0.0 and 1.0. The `accuracy` parameter (default: 10000) is
+ a positive integer literal which controls approximation accuracy at the cost of memory.
+ Higher value of `accuracy` yield better accuracy, `1.0/accuracy` is the relative error of
+ the approximation.
+ """)
+case class ApproximatePercentile(
+ child: Expression,
+ percentageExpression: Expression,
+ accuracyExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int) extends TypedImperativeAggregate[PercentileDigest] {
+
+ def this(child: Expression, percentageExpression: Expression, accuracyExpression: Expression) = {
+ this(child, percentageExpression, accuracyExpression, 0, 0)
+ }
+
+ def this(child: Expression, percentageExpression: Expression) = {
+ this(child, percentageExpression, Literal(ApproximatePercentile.DEFAULT_PERCENTILE_ACCURACY))
+ }
+
+ // Mark as lazy so that accuracyExpression is not evaluated during tree transformation.
+ private lazy val accuracy: Int = accuracyExpression.eval().asInstanceOf[Int]
+
+ private lazy val serializer: PercentileDigestSerializer = new PercentileDigestSerializer
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(DoubleType, TypeCollection(DoubleType, ArrayType), IntegerType)
+ }
+
+ // Mark as lazy so that percentageExpression is not evaluated during tree transformation.
+ private lazy val (returnPercentileArray: Boolean, percentages: Array[Double]) = {
+ (percentageExpression.dataType, percentageExpression.eval()) match {
+ // Rule ImplicitTypeCasts can cast other numeric types to double
+ case (_, num: Double) => (false, Array(num))
+ case (ArrayType(baseType: NumericType, _), arrayData: ArrayData) =>
+ val numericArray = arrayData.toArray(baseType)(baseType.classTag)
+ (true, numericArray.map(baseType.numeric.toDouble))
+ case other =>
+ throw new AnalysisException(s"Invalid data type ${other._1} for parameter percentage")
+ }
+ }
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (accuracy <= 0) {
+ TypeCheckFailure(
+ s"The accuracy provided must be a positive integer literal (current value = $accuracy)")
+ } else if (percentages.exists(percentage => percentage < 0.0D || percentage > 1.0D)) {
+ TypeCheckFailure(
+ s"All percentage values must be between 0.0 and 1.0 " +
+ s"(current = ${percentages.mkString(", ")})")
+ } else {
+ TypeCheckSuccess
+ }
+ }
+
+ override def createAggregationBuffer(): PercentileDigest = {
+ val relativeError = 1.0D / accuracy
+ new PercentileDigest(relativeError)
+ }
+
+ override def update(buffer: PercentileDigest, inputRow: InternalRow): Unit = {
+ val value = child.eval(inputRow)
+ // Ignore empty rows, for example: percentile_approx(null)
+ if (value != null) {
+ buffer.add(value.asInstanceOf[Double])
+ }
+ }
+
+ override def merge(buffer: PercentileDigest, other: PercentileDigest): Unit = {
+ buffer.merge(other)
+ }
+
+ override def eval(buffer: PercentileDigest): Any = {
+ val result = buffer.getPercentiles(percentages)
+ if (result.length == 0) {
+ null
+ } else if (returnPercentileArray) {
+ new GenericArrayData(result)
+ } else {
+ result(0)
+ }
+ }
+
+ override def withNewMutableAggBufferOffset(newOffset: Int): ApproximatePercentile =
+ copy(mutableAggBufferOffset = newOffset)
+
+ override def withNewInputAggBufferOffset(newOffset: Int): ApproximatePercentile =
+ copy(inputAggBufferOffset = newOffset)
+
+ override def children: Seq[Expression] = Seq(child, percentageExpression, accuracyExpression)
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = {
+ if (returnPercentileArray) ArrayType(DoubleType) else DoubleType
+ }
+
+ override def prettyName: String = "percentile_approx"
+
+ override def serialize(obj: PercentileDigest): Array[Byte] = {
+ serializer.serialize(obj)
+ }
+
+ override def deserialize(bytes: Array[Byte]): PercentileDigest = {
+ serializer.deserialize(bytes)
+ }
+}
+
+object ApproximatePercentile {
+
+ // Default accuracy of Percentile approximation. Larger value means better accuracy.
+ // The default relative error can be deduced by defaultError = 1.0 / DEFAULT_PERCENTILE_ACCURACY
+ val DEFAULT_PERCENTILE_ACCURACY: Int = 10000
+
+ /**
+ * PercentileDigest is a probabilistic data structure used for approximating percentiles
+ * with limited memory. PercentileDigest is backed by [[QuantileSummaries]].
+ *
+ * @param summaries underlying probabilistic data structure [[QuantileSummaries]].
+ * @param isCompressed An internal flag from class [[QuantileSummaries]] to indicate whether the
+ * underlying quantileSummaries is compressed.
+ */
+ class PercentileDigest(
+ private var summaries: QuantileSummaries,
+ private var isCompressed: Boolean) {
+
+ // Trigger compression if the QuantileSummaries's buffer length exceeds
+ // compressThresHoldBufferLength. The buffer length can be get by
+ // quantileSummaries.sampled.length
+ private[this] final val compressThresHoldBufferLength: Int = {
+ // Max buffer length after compression.
+ val maxBufferLengthAfterCompression: Int = (1 / summaries.relativeError).toInt * 2
+ // A safe upper bound for buffer length before compression
+ maxBufferLengthAfterCompression * 2
+ }
+
+ def this(relativeError: Double) = {
+ this(new QuantileSummaries(defaultCompressThreshold, relativeError), isCompressed = true)
+ }
+
+ /** Returns compressed object of [[QuantileSummaries]] */
+ def quantileSummaries: QuantileSummaries = {
+ if (!isCompressed) compress()
+ summaries
+ }
+
+ /** Insert an observation value into the PercentileDigest data structure. */
+ def add(value: Double): Unit = {
+ summaries = summaries.insert(value)
+ // The result of QuantileSummaries.insert is un-compressed
+ isCompressed = false
+
+ // Currently, QuantileSummaries ignores the construction parameter compressThresHold,
+ // which may cause QuantileSummaries to occupy unbounded memory. We have to hack around here
+ // to make sure QuantileSummaries doesn't occupy infinite memory.
+ // TODO: Figure out why QuantileSummaries ignores construction parameter compressThresHold
+ if (summaries.sampled.length >= compressThresHoldBufferLength) compress()
+ }
+
+ /** In-place merges in another PercentileDigest. */
+ def merge(other: PercentileDigest): Unit = {
+ if (!isCompressed) compress()
+ summaries = summaries.merge(other.quantileSummaries)
+ }
+
+ /**
+ * Returns the approximate percentiles of all observation values at the given percentages.
+ * A percentile is a watermark value below which a given percentage of observation values fall.
+ * For example, the following code returns the 25th, median, and 75th percentiles of
+ * all observation values:
+ *
+ * {{{
+ * val Array(p25, median, p75) = percentileDigest.getPercentiles(Array(0.25, 0.5, 0.75))
+ * }}}
+ */
+ def getPercentiles(percentages: Array[Double]): Array[Double] = {
+ if (!isCompressed) compress()
+ if (summaries.count == 0 || percentages.length == 0) {
+ Array.empty[Double]
+ } else {
+ val result = new Array[Double](percentages.length)
+ var i = 0
+ while (i < percentages.length) {
+ result(i) = summaries.query(percentages(i))
+ i += 1
+ }
+ result
+ }
+ }
+
+ private final def compress(): Unit = {
+ summaries = summaries.compress()
+ isCompressed = true
+ }
+ }
+
+ /**
+ * Serializer for class [[PercentileDigest]]
+ *
+ * This class is NOT thread safe because usage of ExpressionEncoder is not threadsafe.
+ */
+ class PercentileDigestSerializer {
+
+ // In Scala 2.10, the creation of TypeTag is not thread safe. We need to use ScalaReflectionLock
+ // to protect the creation of this encoder. See SI-6240 for more details.
+ private[this] final val serializer = ScalaReflectionLock.synchronized {
+ ExpressionEncoder[QuantileSummariesData].resolveAndBind()
+ }
+
+ // There are 4 fields in QuantileSummariesData
+ private[this] final val row = new UnsafeRow(4)
+
+ final def serialize(obj: PercentileDigest): Array[Byte] = {
+ val data = new QuantileSummariesData(obj.quantileSummaries)
+ serializer.toRow(data).asInstanceOf[UnsafeRow].getBytes
+ }
+
+ final def deserialize(bytes: Array[Byte]): PercentileDigest = {
+ row.pointTo(bytes, bytes.length)
+ val quantileSummaries = serializer.fromRow(row).toQuantileSummaries
+ new PercentileDigest(quantileSummaries, isCompressed = true)
+ }
+ }
+
+ // An case class to wrap fields of QuantileSummaries, so that we can use the expression encoder
+ // to serialize it.
+ case class QuantileSummariesData(
--- End diff --
QuantileSummaries contains some private var fields, so I think it is better to define a wrapper class instead of changing QuantileSummaries to case class (which is not supposed to contain var fields).
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14868: [SPARK-16283][SQL] Implements percentile_approx a...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/spark/pull/14868
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: Implements percentile_approx aggregation function which ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14868
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/64602/
Test FAILed.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: [SPARK-16283][SQL][WIP] Implements percentile_approx agg...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14868
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/64696/
Test PASSed.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: Implements percentile_approx aggregation function which ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14868
Merged build finished. Test FAILed.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14868: Implements percentile_approx aggregation function which ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14868
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/64642/
Test FAILed.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org