You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by gczsjdy <gi...@git.apache.org> on 2017/03/20 07:35:51 UTC
[GitHub] spark pull request #17359: Add aggreagate expression nGrams
GitHub user gczsjdy opened a pull request:
https://github.com/apache/spark/pull/17359
Add aggreagate expression nGrams
## What changes were proposed in this pull request?
This is the implementation of Hive's `ngrams`, which is a popular statistical and data mining function.
N-grams are subsequences of length N drawn from a longer sequence. The purpose of the `ngrams` aggregate expression is to find the k most frequent n-grams from one or more sequences.
This expression has the pattern of : ngrams(children: Array<Array<String>>(or Array<String>), n: Int, k: Int, accuracy: Int), all parameters are of `Expresion` type, I use the above way to indicate the type after evaluation. Among the parameters,
children indicates the 'longer sequence' we collect n-grams from;
n & k indicate the top k n-grams;
accuracy is related to the memory used for frequency estimation, more memory will give more accurate frequency counts.
## How was this patch tested?
Unit tests are in `NGramsQuerySuite`
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/gczsjdy/spark UdafNGram
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/17359.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 #17359
----
----
---
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 #17359: [SPARK-20028][SQL] Add aggreagate expression nGra...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/17359#discussion_r107351551
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/NGrams.scala ---
@@ -0,0 +1,258 @@
+/*
+ * 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 java.util.HashMap
+
+import org.apache.spark.serializer.KryoSerializer
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult._
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription, ImplicitCastInputTypes, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.NGrams.NGramBuffer
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData}
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Return the top-k n-grams in rows that consist of sequences of strings.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(expr, n, k, accuracy) - Estimates the top-k n-grams in rows that consist of sequences
+ of strings, represented as arrays of strings, or arrays of arrays of strings. 'accuracy' is an
+ optional precision factor that controls memory usage.
+ The parameter 'n' specifies what type of n-grams are being estimated. Unigrams are n = 1, and
+ bigrams are n = 2. Generally, n will not be greater than about 5. The 'k' parameter specifies
+ how many of the highest-frequency n-grams will be returned by the UDAF. The optional precision
+ factor 'accuracy' specifies how much memory to use for estimation; more memory will give
+ more accurate frequency counts, but could crash the JVM. The value will be the max between
+ 'accuracy'(0 if it's not specified) and 1000/k, which indicates the max number of n-grams
+ which are kept in the internal HashMap.
+ The output is an array of maps with the top-k n-grams and corresponding frequency.
+ """,
+ extended = """
+ Examples:
+ > SELECT ngrams(array("abc", "abc", "bcd", "abc", "bcd"), 2, 4);
+ [{["abc","bcd"]:2.0},
+ {["abc","abc"]:1.0},
+ {["bcd","abc"]:1.0}]
+ """)
+case class NGrams(
+ child: Expression,
+ nExpression: Expression,
+ kExpression: Expression,
+ accuracyExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int)
+ extends TypedImperativeAggregate[NGramBuffer] with ImplicitCastInputTypes {
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression,
+ accuracyExpression: Expression) = {
+ this(child, nExpression, kExpression, NGrams.getAccuracy(kExpression, accuracyExpression), 0, 0)
+ }
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression) = {
+ this(child, nExpression, kExpression, Literal(0))
+ }
+
+ private lazy val n: Int = nExpression.eval().asInstanceOf[Int]
+ private lazy val k: Int = kExpression.eval().asInstanceOf[Int]
+ private lazy val accuracy: Int = accuracyExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(ArrayType(StringType, false), ArrayType(ArrayType(StringType, false))),
+ IntegerType, IntegerType, IntegerType)
+ }
+
+ val isArrayOfString = child.dataType == ArrayType(StringType, false) ||
+ child.dataType == ArrayType(StringType, true)
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (!nExpression.foldable || !kExpression.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 {
+ TypeCheckSuccess
+ }
+ }
+
+ override def createAggregationBuffer(): NGramBuffer = {
+ new NGramBuffer(n, k, accuracy, new HashMap[Vector[UTF8String], Double]())
+ }
+
+ def updateArray(genericArrayData: GenericArrayData, buffer: NGramBuffer, inputRow: InternalRow) {
+ val values = (0 until genericArrayData.numElements()).map(genericArrayData.get(_, StringType).
+ asInstanceOf[UTF8String]).toVector
+ val nGrams = getNGrams(values, n)
+ nGrams.foreach(buffer.add(_))
+ buffer.trim()
+ }
+ override def update(buffer: NGramBuffer, inputRow: InternalRow): NGramBuffer = {
+ if (isArrayOfString) {
+ updateArray(child.eval(inputRow).asInstanceOf[GenericArrayData], buffer, inputRow)
+ }
+ else {
+ val arrayOfArray = child.eval(inputRow).asInstanceOf[GenericArrayData]
+ for (i <- 0 until arrayOfArray.numElements()) {
+ updateArray(arrayOfArray.getArray(i).asInstanceOf[GenericArrayData], buffer, inputRow)
+ }
+ }
+ buffer
+ }
+
+ override def merge(buffer: NGramBuffer, input: NGramBuffer): NGramBuffer = {
+ buffer.merge(input)
+ buffer.trim()
+ buffer
+ }
+
+ override def eval(buffer: NGramBuffer): Any = {
+ val topKNGrams = buffer.getTopKNGrams().map((keyValuePair: (Vector[UTF8String], Double)) => {
+ val arrayKey = new GenericArrayData(Vector(new GenericArrayData(keyValuePair._1)))
+ val arrayValue = new GenericArrayData(Vector(keyValuePair._2))
+ new ArrayBasedMapData(arrayKey, arrayValue)
+ }).toVector
+ new GenericArrayData(topKNGrams)
+ }
+
+ private def getNGrams(values: Vector[UTF8String], n: Int): Vector[Vector[UTF8String]] = {
+ if (values.length >= n) {
+ values.sliding(n).toVector
+ }
+ else {
+ Vector()
+ }
+ }
+
+ override def withNewMutableAggBufferOffset(newOffset: Int): NGrams =
+ copy(mutableAggBufferOffset = newOffset)
+
+ override def withNewInputAggBufferOffset(newOffset: Int): NGrams =
+ copy(inputAggBufferOffset = newOffset)
+
+ override def children: Seq[Expression] = Seq(child, nExpression, kExpression, accuracyExpression)
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = ArrayType(MapType(ArrayType(StringType), DoubleType))
+
+ override def prettyName: String = "ngrams"
+
+ override def serialize(obj: NGramBuffer): Array[Byte] = {
+ NGrams.serializer.serialize(obj)
+ }
+
+ override def deserialize(bytes: Array[Byte]): NGramBuffer = {
+ NGrams.serializer.deserialize(bytes)
+ }
+}
+
+object NGrams {
+
+ val DEFAULT_ACCURACY: Int = 1000
+
+ private def getAccuracy(kExpression: Expression, accuracyExpression: Expression): Expression = {
+ val accuracy = accuracyExpression.eval().asInstanceOf[Int]
+ val k = kExpression.eval().asInstanceOf[Int]
+ Literal(accuracy.max(DEFAULT_ACCURACY / k))
+ }
+
+ val kryoSerializer: KryoSerializer = new KryoSerializer(new SparkConf())
+
+ import collection.JavaConverters._
+
+ class NGramBuffer(val n: Int,
+ val k: Int,
+ val precisionFactor: Int,
+ val frequencyMap: HashMap[Vector[UTF8String], Double]) {
+ def add(ng: Vector[UTF8String]): Unit = {
+ var currentFrequency: Double = frequencyMap.get(ng)
+ if (currentFrequency == null.asInstanceOf[Double]) {
+ currentFrequency = 1.0D
+ } else {
+ currentFrequency += 1
+ }
+ frequencyMap.put(ng, currentFrequency)
+ }
+
+ def merge(other: NGramBuffer): Unit = {
+ other.frequencyMap.asScala.foreach((keyValuePair: (Vector[UTF8String], Double)) => {
+ val key = keyValuePair._1
+ val value = keyValuePair._2
+ val originalValue = frequencyMap.getOrDefault(key, 0.0D)
+ frequencyMap.put(key, originalValue + value)
+ })
+ }
+
+ def sortWithTwoFields(frequencyDescend: Boolean)
+ (keyWithFrequency: Tuple2[Vector[UTF8String], Double],
+ keyWithFrequency2: Tuple2[Vector[UTF8String], Double]): Boolean = {
+ if (keyWithFrequency._2 != keyWithFrequency2._2) {
+ (keyWithFrequency._2 < keyWithFrequency2._2) ^ frequencyDescend
+ }
+ else {
+ val keyVector = keyWithFrequency._1
+ val keyVector2 = keyWithFrequency2._1
+ for (i <- 0 until keyVector.length) {
+ if (keyVector(i) != keyVector2(i)) {
+ return (keyVector(i).compare(keyVector2(i))) < 0
+ }
+ }
+ true
+ }
+ }
+
+ def trim(): Unit = {
+ if (frequencyMap.size() > 2 * k * precisionFactor) {
+ val orderedWithIndex = frequencyMap.asScala.iterator.toVector.
+ sortWith(sortWithTwoFields(frequencyDescend = false)).zipWithIndex
+ orderedWithIndex.takeWhile(_._2 < frequencyMap.size() - k * precisionFactor).map(_._1).
+ foreach(keyValuePair => frequencyMap.remove(keyValuePair._1))
+ }
+ }
+
+ def getTopKNGrams(): Seq[(Vector[UTF8String], Double)] = {
+ frequencyMap.asScala.iterator.toVector.sortWith(sortWithTwoFields(frequencyDescend = true)).
+ zipWithIndex.takeWhile(_._2 < k).map(_._1)
+ }
+
+ }
+
+ class NGramBufferSerializer {
+
+ final def serialize(obj: NGramBuffer): Array[Byte] = {
+ NGrams.kryoSerializer.newInstance().serialize[NGramBuffer](obj).array()
--- End diff --
Do we really need to `newInstance` every time `serialize` and `deserialize`?
---
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 #17359: [SPARK-20028][SQL] Add aggreagate expression nGrams
Posted by gczsjdy <gi...@git.apache.org>.
Github user gczsjdy commented on the issue:
https://github.com/apache/spark/pull/17359
@viirya Sorry for the late reply. It seems Spark cannot use Hive `GenericUDAFnGrams`, since Spark only supports subclass of `AbstractGenericUDAFResolver` & `UDAF` for Hive UDAF, while `GenericUDAFnGrams` is subclass of `GenericUDAFResolver`. Spark will return a `No handler for Hive UDF` exception. So this may be a good reason to add new feature for Spark aggregate expression.
Or do you think it's reasonable to make Spark support registering subclass of `GenericUDAFResolver`?
---
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 #17359: [SPARK-20028][SQL] Add aggreagate expression nGra...
Posted by gczsjdy <gi...@git.apache.org>.
Github user gczsjdy commented on a diff in the pull request:
https://github.com/apache/spark/pull/17359#discussion_r107428643
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/NGrams.scala ---
@@ -0,0 +1,258 @@
+/*
+ * 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 java.util.HashMap
+
+import org.apache.spark.serializer.KryoSerializer
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult._
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription, ImplicitCastInputTypes, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.NGrams.NGramBuffer
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData}
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Return the top-k n-grams in rows that consist of sequences of strings.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(expr, n, k, accuracy) - Estimates the top-k n-grams in rows that consist of sequences
+ of strings, represented as arrays of strings, or arrays of arrays of strings. 'accuracy' is an
+ optional precision factor that controls memory usage.
+ The parameter 'n' specifies what type of n-grams are being estimated. Unigrams are n = 1, and
+ bigrams are n = 2. Generally, n will not be greater than about 5. The 'k' parameter specifies
+ how many of the highest-frequency n-grams will be returned by the UDAF. The optional precision
+ factor 'accuracy' specifies how much memory to use for estimation; more memory will give
+ more accurate frequency counts, but could crash the JVM. The value will be the max between
+ 'accuracy'(0 if it's not specified) and 1000/k, which indicates the max number of n-grams
+ which are kept in the internal HashMap.
+ The output is an array of maps with the top-k n-grams and corresponding frequency.
+ """,
+ extended = """
+ Examples:
+ > SELECT ngrams(array("abc", "abc", "bcd", "abc", "bcd"), 2, 4);
+ [{["abc","bcd"]:2.0},
+ {["abc","abc"]:1.0},
+ {["bcd","abc"]:1.0}]
+ """)
+case class NGrams(
+ child: Expression,
+ nExpression: Expression,
+ kExpression: Expression,
+ accuracyExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int)
+ extends TypedImperativeAggregate[NGramBuffer] with ImplicitCastInputTypes {
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression,
+ accuracyExpression: Expression) = {
+ this(child, nExpression, kExpression, NGrams.getAccuracy(kExpression, accuracyExpression), 0, 0)
+ }
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression) = {
+ this(child, nExpression, kExpression, Literal(0))
+ }
+
+ private lazy val n: Int = nExpression.eval().asInstanceOf[Int]
--- End diff --
Right, thanks
---
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 #17359: [SPARK-20028][SQL] Add aggreagate expression nGrams
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on the issue:
https://github.com/apache/spark/pull/17359
cc @wzhfy @viirya Are you interested in reviewing this PR?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #17359: [SPARK-20028][SQL] Add aggreagate expression nGra...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/17359#discussion_r107347431
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/NGrams.scala ---
@@ -0,0 +1,258 @@
+/*
+ * 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 java.util.HashMap
+
+import org.apache.spark.serializer.KryoSerializer
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult._
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription, ImplicitCastInputTypes, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.NGrams.NGramBuffer
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData}
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Return the top-k n-grams in rows that consist of sequences of strings.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(expr, n, k, accuracy) - Estimates the top-k n-grams in rows that consist of sequences
+ of strings, represented as arrays of strings, or arrays of arrays of strings. 'accuracy' is an
+ optional precision factor that controls memory usage.
+ The parameter 'n' specifies what type of n-grams are being estimated. Unigrams are n = 1, and
+ bigrams are n = 2. Generally, n will not be greater than about 5. The 'k' parameter specifies
+ how many of the highest-frequency n-grams will be returned by the UDAF. The optional precision
+ factor 'accuracy' specifies how much memory to use for estimation; more memory will give
+ more accurate frequency counts, but could crash the JVM. The value will be the max between
+ 'accuracy'(0 if it's not specified) and 1000/k, which indicates the max number of n-grams
+ which are kept in the internal HashMap.
+ The output is an array of maps with the top-k n-grams and corresponding frequency.
+ """,
+ extended = """
+ Examples:
+ > SELECT ngrams(array("abc", "abc", "bcd", "abc", "bcd"), 2, 4);
+ [{["abc","bcd"]:2.0},
+ {["abc","abc"]:1.0},
+ {["bcd","abc"]:1.0}]
+ """)
+case class NGrams(
+ child: Expression,
+ nExpression: Expression,
+ kExpression: Expression,
+ accuracyExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int)
+ extends TypedImperativeAggregate[NGramBuffer] with ImplicitCastInputTypes {
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression,
+ accuracyExpression: Expression) = {
+ this(child, nExpression, kExpression, NGrams.getAccuracy(kExpression, accuracyExpression), 0, 0)
+ }
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression) = {
+ this(child, nExpression, kExpression, Literal(0))
+ }
+
+ private lazy val n: Int = nExpression.eval().asInstanceOf[Int]
+ private lazy val k: Int = kExpression.eval().asInstanceOf[Int]
+ private lazy val accuracy: Int = accuracyExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(ArrayType(StringType, false), ArrayType(ArrayType(StringType, false))),
+ IntegerType, IntegerType, IntegerType)
+ }
+
+ val isArrayOfString = child.dataType == ArrayType(StringType, false) ||
+ child.dataType == ArrayType(StringType, true)
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (!nExpression.foldable || !kExpression.foldable || !accuracyExpression.foldable) {
+ TypeCheckFailure(s"The accuracy or percentage provided must be a constant literal")
--- End diff --
`must be a foldable integer expression`.
---
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 #17359: [SPARK-20028][SQL] Add aggreagate expression nGrams
Posted by gczsjdy <gi...@git.apache.org>.
Github user gczsjdy commented on the issue:
https://github.com/apache/spark/pull/17359
@rxin @cloud-fan @gatorsmile @viirya @tejasapatil Could you please help me review this PR? Or is there anything I can do on this work?
---
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 #17359: [SPARK-20028][SQL] Add aggreagate expression nGra...
Posted by gczsjdy <gi...@git.apache.org>.
Github user gczsjdy commented on a diff in the pull request:
https://github.com/apache/spark/pull/17359#discussion_r107415412
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/NGrams.scala ---
@@ -0,0 +1,258 @@
+/*
+ * 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 java.util.HashMap
+
+import org.apache.spark.serializer.KryoSerializer
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult._
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription, ImplicitCastInputTypes, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.NGrams.NGramBuffer
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData}
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Return the top-k n-grams in rows that consist of sequences of strings.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(expr, n, k, accuracy) - Estimates the top-k n-grams in rows that consist of sequences
+ of strings, represented as arrays of strings, or arrays of arrays of strings. 'accuracy' is an
+ optional precision factor that controls memory usage.
+ The parameter 'n' specifies what type of n-grams are being estimated. Unigrams are n = 1, and
+ bigrams are n = 2. Generally, n will not be greater than about 5. The 'k' parameter specifies
+ how many of the highest-frequency n-grams will be returned by the UDAF. The optional precision
+ factor 'accuracy' specifies how much memory to use for estimation; more memory will give
+ more accurate frequency counts, but could crash the JVM. The value will be the max between
+ 'accuracy'(0 if it's not specified) and 1000/k, which indicates the max number of n-grams
+ which are kept in the internal HashMap.
+ The output is an array of maps with the top-k n-grams and corresponding frequency.
+ """,
+ extended = """
+ Examples:
+ > SELECT ngrams(array("abc", "abc", "bcd", "abc", "bcd"), 2, 4);
+ [{["abc","bcd"]:2.0},
+ {["abc","abc"]:1.0},
+ {["bcd","abc"]:1.0}]
+ """)
+case class NGrams(
+ child: Expression,
+ nExpression: Expression,
+ kExpression: Expression,
+ accuracyExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int)
+ extends TypedImperativeAggregate[NGramBuffer] with ImplicitCastInputTypes {
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression,
+ accuracyExpression: Expression) = {
+ this(child, nExpression, kExpression, NGrams.getAccuracy(kExpression, accuracyExpression), 0, 0)
+ }
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression) = {
+ this(child, nExpression, kExpression, Literal(0))
+ }
+
+ private lazy val n: Int = nExpression.eval().asInstanceOf[Int]
+ private lazy val k: Int = kExpression.eval().asInstanceOf[Int]
+ private lazy val accuracy: Int = accuracyExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(ArrayType(StringType, false), ArrayType(ArrayType(StringType, false))),
+ IntegerType, IntegerType, IntegerType)
+ }
+
+ val isArrayOfString = child.dataType == ArrayType(StringType, false) ||
+ child.dataType == ArrayType(StringType, true)
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (!nExpression.foldable || !kExpression.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 {
+ TypeCheckSuccess
+ }
+ }
+
+ override def createAggregationBuffer(): NGramBuffer = {
+ new NGramBuffer(n, k, accuracy, new HashMap[Vector[UTF8String], Double]())
+ }
+
+ def updateArray(genericArrayData: GenericArrayData, buffer: NGramBuffer, inputRow: InternalRow) {
+ val values = (0 until genericArrayData.numElements()).map(genericArrayData.get(_, StringType).
+ asInstanceOf[UTF8String]).toVector
+ val nGrams = getNGrams(values, n)
+ nGrams.foreach(buffer.add(_))
+ buffer.trim()
+ }
+ override def update(buffer: NGramBuffer, inputRow: InternalRow): NGramBuffer = {
+ if (isArrayOfString) {
+ updateArray(child.eval(inputRow).asInstanceOf[GenericArrayData], buffer, inputRow)
+ }
+ else {
+ val arrayOfArray = child.eval(inputRow).asInstanceOf[GenericArrayData]
+ for (i <- 0 until arrayOfArray.numElements()) {
+ updateArray(arrayOfArray.getArray(i).asInstanceOf[GenericArrayData], buffer, inputRow)
+ }
+ }
+ buffer
+ }
+
+ override def merge(buffer: NGramBuffer, input: NGramBuffer): NGramBuffer = {
+ buffer.merge(input)
+ buffer.trim()
+ buffer
+ }
+
+ override def eval(buffer: NGramBuffer): Any = {
+ val topKNGrams = buffer.getTopKNGrams().map((keyValuePair: (Vector[UTF8String], Double)) => {
+ val arrayKey = new GenericArrayData(Vector(new GenericArrayData(keyValuePair._1)))
+ val arrayValue = new GenericArrayData(Vector(keyValuePair._2))
+ new ArrayBasedMapData(arrayKey, arrayValue)
+ }).toVector
+ new GenericArrayData(topKNGrams)
--- End diff --
My fault, it's weird, I run it and it tells me `Scalastyle checks 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 #17359: [SPARK-20028][SQL] Add aggreagate expression nGra...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/17359#discussion_r107344830
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/NGrams.scala ---
@@ -0,0 +1,258 @@
+/*
+ * 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 java.util.HashMap
+
+import org.apache.spark.serializer.KryoSerializer
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult._
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription, ImplicitCastInputTypes, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.NGrams.NGramBuffer
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData}
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Return the top-k n-grams in rows that consist of sequences of strings.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(expr, n, k, accuracy) - Estimates the top-k n-grams in rows that consist of sequences
+ of strings, represented as arrays of strings, or arrays of arrays of strings. 'accuracy' is an
--- End diff --
does Hive also support `arrays of arrays of strings`?
---
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 #17359: [SPARK-20028][SQL] Add aggreagate expression nGrams
Posted by gczsjdy <gi...@git.apache.org>.
Github user gczsjdy commented on the issue:
https://github.com/apache/spark/pull/17359
@rxin My fault, the example I gave is far from practical use and I have updated. Actually, we can use it whenever the text analysis is reasonable to be based on frequencies of word sequences. For example, analyze movie or product reviews, find common sequences of URL accesses, etc.
---
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 #17359: [SPARK-20028][SQL] Add aggreagate expression nGra...
Posted by gczsjdy <gi...@git.apache.org>.
Github user gczsjdy commented on a diff in the pull request:
https://github.com/apache/spark/pull/17359#discussion_r107429936
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/NGrams.scala ---
@@ -0,0 +1,258 @@
+/*
+ * 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 java.util.HashMap
+
+import org.apache.spark.serializer.KryoSerializer
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult._
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription, ImplicitCastInputTypes, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.NGrams.NGramBuffer
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData}
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Return the top-k n-grams in rows that consist of sequences of strings.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(expr, n, k, accuracy) - Estimates the top-k n-grams in rows that consist of sequences
+ of strings, represented as arrays of strings, or arrays of arrays of strings. 'accuracy' is an
+ optional precision factor that controls memory usage.
+ The parameter 'n' specifies what type of n-grams are being estimated. Unigrams are n = 1, and
+ bigrams are n = 2. Generally, n will not be greater than about 5. The 'k' parameter specifies
+ how many of the highest-frequency n-grams will be returned by the UDAF. The optional precision
+ factor 'accuracy' specifies how much memory to use for estimation; more memory will give
+ more accurate frequency counts, but could crash the JVM. The value will be the max between
+ 'accuracy'(0 if it's not specified) and 1000/k, which indicates the max number of n-grams
+ which are kept in the internal HashMap.
+ The output is an array of maps with the top-k n-grams and corresponding frequency.
+ """,
+ extended = """
+ Examples:
+ > SELECT ngrams(array("abc", "abc", "bcd", "abc", "bcd"), 2, 4);
+ [{["abc","bcd"]:2.0},
+ {["abc","abc"]:1.0},
+ {["bcd","abc"]:1.0}]
+ """)
+case class NGrams(
+ child: Expression,
+ nExpression: Expression,
+ kExpression: Expression,
+ accuracyExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int)
+ extends TypedImperativeAggregate[NGramBuffer] with ImplicitCastInputTypes {
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression,
+ accuracyExpression: Expression) = {
+ this(child, nExpression, kExpression, NGrams.getAccuracy(kExpression, accuracyExpression), 0, 0)
+ }
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression) = {
+ this(child, nExpression, kExpression, Literal(0))
+ }
+
+ private lazy val n: Int = nExpression.eval().asInstanceOf[Int]
+ private lazy val k: Int = kExpression.eval().asInstanceOf[Int]
+ private lazy val accuracy: Int = accuracyExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(ArrayType(StringType, false), ArrayType(ArrayType(StringType, false))),
+ IntegerType, IntegerType, IntegerType)
+ }
+
+ val isArrayOfString = child.dataType == ArrayType(StringType, false) ||
+ child.dataType == ArrayType(StringType, true)
--- End diff --
To add, Array of String with null will cause runtime exceptions.
---
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 #17359: [SPARK-20028][SQL] Add aggreagate expression nGrams
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on the issue:
https://github.com/apache/spark/pull/17359
Why do we want this? Seems extremely low usage on this function in the wild.
---
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 #17359: [SPARK-20028][SQL] Add aggreagate expression nGra...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/17359#discussion_r107353470
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/NGrams.scala ---
@@ -0,0 +1,258 @@
+/*
+ * 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 java.util.HashMap
+
+import org.apache.spark.serializer.KryoSerializer
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult._
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription, ImplicitCastInputTypes, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.NGrams.NGramBuffer
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData}
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Return the top-k n-grams in rows that consist of sequences of strings.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(expr, n, k, accuracy) - Estimates the top-k n-grams in rows that consist of sequences
+ of strings, represented as arrays of strings, or arrays of arrays of strings. 'accuracy' is an
+ optional precision factor that controls memory usage.
+ The parameter 'n' specifies what type of n-grams are being estimated. Unigrams are n = 1, and
+ bigrams are n = 2. Generally, n will not be greater than about 5. The 'k' parameter specifies
+ how many of the highest-frequency n-grams will be returned by the UDAF. The optional precision
+ factor 'accuracy' specifies how much memory to use for estimation; more memory will give
+ more accurate frequency counts, but could crash the JVM. The value will be the max between
+ 'accuracy'(0 if it's not specified) and 1000/k, which indicates the max number of n-grams
+ which are kept in the internal HashMap.
+ The output is an array of maps with the top-k n-grams and corresponding frequency.
+ """,
+ extended = """
+ Examples:
+ > SELECT ngrams(array("abc", "abc", "bcd", "abc", "bcd"), 2, 4);
+ [{["abc","bcd"]:2.0},
+ {["abc","abc"]:1.0},
+ {["bcd","abc"]:1.0}]
+ """)
+case class NGrams(
+ child: Expression,
+ nExpression: Expression,
+ kExpression: Expression,
+ accuracyExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int)
+ extends TypedImperativeAggregate[NGramBuffer] with ImplicitCastInputTypes {
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression,
+ accuracyExpression: Expression) = {
+ this(child, nExpression, kExpression, NGrams.getAccuracy(kExpression, accuracyExpression), 0, 0)
+ }
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression) = {
+ this(child, nExpression, kExpression, Literal(0))
+ }
+
+ private lazy val n: Int = nExpression.eval().asInstanceOf[Int]
--- End diff --
also k.
---
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 #17359: [SPARK-20028][SQL] Add aggreagate expression nGrams
Posted by gczsjdy <gi...@git.apache.org>.
Github user gczsjdy commented on the issue:
https://github.com/apache/spark/pull/17359
Sorry, but I think this is inactive. Thanks for your attention. @wzhfy @viirya @gatorsmile
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #17359: [SPARK-20028][SQL] Add aggreagate expression nGra...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/17359#discussion_r107343838
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/NGrams.scala ---
@@ -0,0 +1,258 @@
+/*
+ * 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 java.util.HashMap
+
+import org.apache.spark.serializer.KryoSerializer
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult._
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription, ImplicitCastInputTypes, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.NGrams.NGramBuffer
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData}
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Return the top-k n-grams in rows that consist of sequences of strings.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(expr, n, k, accuracy) - Estimates the top-k n-grams in rows that consist of sequences
+ of strings, represented as arrays of strings, or arrays of arrays of strings. 'accuracy' is an
+ optional precision factor that controls memory usage.
+ The parameter 'n' specifies what type of n-grams are being estimated. Unigrams are n = 1, and
+ bigrams are n = 2. Generally, n will not be greater than about 5. The 'k' parameter specifies
+ how many of the highest-frequency n-grams will be returned by the UDAF. The optional precision
+ factor 'accuracy' specifies how much memory to use for estimation; more memory will give
+ more accurate frequency counts, but could crash the JVM. The value will be the max between
+ 'accuracy'(0 if it's not specified) and 1000/k, which indicates the max number of n-grams
+ which are kept in the internal HashMap.
+ The output is an array of maps with the top-k n-grams and corresponding frequency.
+ """,
+ extended = """
+ Examples:
+ > SELECT ngrams(array('abc', 'abc', 'bcd', 'abc', 'bcd'), 2, 4);
+ [["abc","bcd"]:2.0},
--- End diff --
oh. i see. your unit of unigram is each element in the array, not a character.
---
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 #17359: [SPARK-20028][SQL] Add aggreagate expression nGrams
Posted by wzhfy <gi...@git.apache.org>.
Github user wzhfy commented on the issue:
https://github.com/apache/spark/pull/17359
@gatorsmile Sure, I haven't read the context, but it's been nearly half a year since last update, is this PR still active @gczsjdy ?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #17359: [SPARK-20028][SQL] Add aggreagate expression nGra...
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/17359#discussion_r110568699
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/NGrams.scala ---
@@ -0,0 +1,249 @@
+/*
+ * 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 java.util.HashMap
+
+import org.apache.spark.serializer.KryoSerializer
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult._
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription, ImplicitCastInputTypes, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.NGrams.NGramBuffer
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData}
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Return the top-k n-grams in rows that consist of sequences of strings.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(expr, n, k, accuracy) - Estimates the top-k n-grams in rows that consist of sequences
+ of strings, represented as arrays of strings, or arrays of arrays of strings. 'accuracy' is an
+ optional precision factor that controls memory usage.
+ The parameter 'n' specifies what type of n-grams are being estimated. Unigrams are n = 1, and
+ bigrams are n = 2. Generally, n will not be greater than about 5. The 'k' parameter specifies
+ how many of the highest-frequency n-grams will be returned by the UDAF. The optional precision
+ factor 'accuracy' specifies how much memory to use for estimation; more memory will give
+ more accurate frequency counts, but could crash the JVM. The value will be the max between
+ 'accuracy'(0 if it's not specified) and 1000/k, which indicates the max number of n-grams
+ which are kept in the internal HashMap.
+ The output is an array of maps with the top-k n-grams and corresponding frequency.
+ """,
+ extended = """
+ Examples:
+ > SELECT ngrams(array("abc", "abc", "bcd", "abc", "bcd"), 2, 4);
+ [{["abc","bcd"]:2.0},
+ {["abc","abc"]:1.0},
+ {["bcd","abc"]:1.0}]
+ """)
+case class NGrams(
+ child: Expression,
+ nExpression: Expression,
+ kExpression: Expression,
+ accuracyExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int)
+ extends TypedImperativeAggregate[NGramBuffer] with ImplicitCastInputTypes {
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression,
+ accuracyExpression: Expression) = {
+ this(child, nExpression, kExpression, NGrams.getAccuracy(kExpression, accuracyExpression), 0, 0)
+ }
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression) = {
+ this(child, nExpression, kExpression, Literal(0))
+ }
+
+ private lazy val n: Int = nExpression.eval().asInstanceOf[Int]
+ private lazy val k: Int = kExpression.eval().asInstanceOf[Int]
+ private lazy val accuracy: Int = accuracyExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(ArrayType(StringType, false), ArrayType(ArrayType(StringType, false))),
+ IntegerType, IntegerType, IntegerType)
+ }
+
+ val isArrayOfString = child.dataType == ArrayType(StringType, false) ||
+ child.dataType == ArrayType(StringType, true)
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (!nExpression.foldable || !kExpression.foldable || !accuracyExpression.foldable) {
+ TypeCheckFailure(s"The accuracy or percentage provided must be a foldable integer expression")
+ } else if (n <= 0) {
+ TypeCheckFailure(s"The n provided must be a positive integer (current value = $n)")
+ } else if (k <= 0) {
+ TypeCheckFailure(s"The k provided must be a positive integer (current value = $k)")
+ } else if (accuracy <= 0) {
+ TypeCheckFailure(
+ s"The accuracy provided must be a positive integer (current value = $accuracy)")
+ } else {
+ TypeCheckSuccess
+ }
+ }
+
+ override def createAggregationBuffer(): NGramBuffer = {
+ new NGramBuffer(n, k, accuracy, new HashMap[Vector[UTF8String], Double]())
+ }
+
+ def updateArray(genericArrayData: GenericArrayData, buffer: NGramBuffer, inputRow: InternalRow) {
+ val values = (0 until genericArrayData.numElements()).map(genericArrayData.get(_, StringType).
+ asInstanceOf[UTF8String]).toVector
+ val nGrams = getNGrams(values, n)
+ nGrams.foreach(buffer.add(_))
+ buffer.trim()
+ }
+ override def update(buffer: NGramBuffer, inputRow: InternalRow): NGramBuffer = {
+ if (isArrayOfString) {
+ updateArray(child.eval(inputRow).asInstanceOf[GenericArrayData], buffer, inputRow)
+ } else {
+ val arrayOfArray = child.eval(inputRow).asInstanceOf[GenericArrayData]
+ for (i <- 0 until arrayOfArray.numElements()) {
+ updateArray(arrayOfArray.getArray(i).asInstanceOf[GenericArrayData], buffer, inputRow)
+ }
+ }
+ buffer
+ }
+
+ override def merge(buffer: NGramBuffer, input: NGramBuffer): NGramBuffer = {
+ buffer.merge(input)
+ buffer.trim()
+ buffer
+ }
+
+ override def eval(buffer: NGramBuffer): Any = {
+ val topKNGrams = buffer.getTopKNGrams().map((keyValuePair: (Vector[UTF8String], Double)) => {
+ val arrayKey = new GenericArrayData(Vector(new GenericArrayData(keyValuePair._1)))
+ val arrayValue = new GenericArrayData(Vector(keyValuePair._2))
+ new ArrayBasedMapData(arrayKey, arrayValue)
+ }).toVector
+ new GenericArrayData(topKNGrams)
+ }
+
+ private def getNGrams(values: Vector[UTF8String], n: Int): Vector[Vector[UTF8String]] = {
+ if (values.length >= n) {
+ values.sliding(n).toVector
+ } else {
+ Vector()
+ }
+ }
+
+ override def withNewMutableAggBufferOffset(newOffset: Int): NGrams =
+ copy(mutableAggBufferOffset = newOffset)
+
+ override def withNewInputAggBufferOffset(newOffset: Int): NGrams =
+ copy(inputAggBufferOffset = newOffset)
+
+ override def children: Seq[Expression] = Seq(child, nExpression, kExpression, accuracyExpression)
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = ArrayType(MapType(ArrayType(StringType), DoubleType))
+
+ override def prettyName: String = "ngrams"
+
+ override def serialize(obj: NGramBuffer): Array[Byte] = {
+ NGrams.serializer.serialize(obj)
+ }
+
+ override def deserialize(bytes: Array[Byte]): NGramBuffer = {
+ NGrams.serializer.deserialize(bytes)
+ }
+}
+
+object NGrams {
+
+ val DEFAULT_ACCURACY: Int = 1000
+
+ private def getAccuracy(kExpression: Expression, accuracyExpression: Expression): Expression = {
+ val accuracy = accuracyExpression.eval().asInstanceOf[Int]
+ val k = kExpression.eval().asInstanceOf[Int]
+ Literal(accuracy.max(DEFAULT_ACCURACY / k))
+ }
+
+ val kryoSerializer: KryoSerializer = new KryoSerializer(new SparkConf())
+
+ import collection.JavaConverters._
+
+ class NGramBuffer(val n: Int,
+ val k: Int,
+ val precisionFactor: Int,
+ val frequencyMap: HashMap[Vector[UTF8String], Double]) {
--- End diff --
why `frequencyMap` need to be a constructor parameter?
---
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 #17359: [SPARK-20028][SQL] Add aggreagate expression nGra...
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/17359#discussion_r110568482
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/NGrams.scala ---
@@ -0,0 +1,249 @@
+/*
+ * 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 java.util.HashMap
+
+import org.apache.spark.serializer.KryoSerializer
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult._
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription, ImplicitCastInputTypes, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.NGrams.NGramBuffer
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData}
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Return the top-k n-grams in rows that consist of sequences of strings.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(expr, n, k, accuracy) - Estimates the top-k n-grams in rows that consist of sequences
+ of strings, represented as arrays of strings, or arrays of arrays of strings. 'accuracy' is an
+ optional precision factor that controls memory usage.
+ The parameter 'n' specifies what type of n-grams are being estimated. Unigrams are n = 1, and
+ bigrams are n = 2. Generally, n will not be greater than about 5. The 'k' parameter specifies
+ how many of the highest-frequency n-grams will be returned by the UDAF. The optional precision
+ factor 'accuracy' specifies how much memory to use for estimation; more memory will give
+ more accurate frequency counts, but could crash the JVM. The value will be the max between
+ 'accuracy'(0 if it's not specified) and 1000/k, which indicates the max number of n-grams
+ which are kept in the internal HashMap.
+ The output is an array of maps with the top-k n-grams and corresponding frequency.
+ """,
+ extended = """
+ Examples:
+ > SELECT ngrams(array("abc", "abc", "bcd", "abc", "bcd"), 2, 4);
+ [{["abc","bcd"]:2.0},
+ {["abc","abc"]:1.0},
+ {["bcd","abc"]:1.0}]
+ """)
+case class NGrams(
+ child: Expression,
+ nExpression: Expression,
+ kExpression: Expression,
+ accuracyExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int)
+ extends TypedImperativeAggregate[NGramBuffer] with ImplicitCastInputTypes {
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression,
+ accuracyExpression: Expression) = {
+ this(child, nExpression, kExpression, NGrams.getAccuracy(kExpression, accuracyExpression), 0, 0)
+ }
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression) = {
+ this(child, nExpression, kExpression, Literal(0))
+ }
+
+ private lazy val n: Int = nExpression.eval().asInstanceOf[Int]
+ private lazy val k: Int = kExpression.eval().asInstanceOf[Int]
+ private lazy val accuracy: Int = accuracyExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(ArrayType(StringType, false), ArrayType(ArrayType(StringType, false))),
--- End diff --
why do we have special requirement about nullability?
---
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 #17359: [SPARK-20028][SQL] Add aggreagate expression nGra...
Posted by gczsjdy <gi...@git.apache.org>.
Github user gczsjdy commented on a diff in the pull request:
https://github.com/apache/spark/pull/17359#discussion_r110820889
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/NGrams.scala ---
@@ -0,0 +1,249 @@
+/*
+ * 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 java.util.HashMap
+
+import org.apache.spark.serializer.KryoSerializer
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult._
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription, ImplicitCastInputTypes, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.NGrams.NGramBuffer
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData}
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Return the top-k n-grams in rows that consist of sequences of strings.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(expr, n, k, accuracy) - Estimates the top-k n-grams in rows that consist of sequences
+ of strings, represented as arrays of strings, or arrays of arrays of strings. 'accuracy' is an
+ optional precision factor that controls memory usage.
+ The parameter 'n' specifies what type of n-grams are being estimated. Unigrams are n = 1, and
+ bigrams are n = 2. Generally, n will not be greater than about 5. The 'k' parameter specifies
+ how many of the highest-frequency n-grams will be returned by the UDAF. The optional precision
+ factor 'accuracy' specifies how much memory to use for estimation; more memory will give
+ more accurate frequency counts, but could crash the JVM. The value will be the max between
+ 'accuracy'(0 if it's not specified) and 1000/k, which indicates the max number of n-grams
+ which are kept in the internal HashMap.
+ The output is an array of maps with the top-k n-grams and corresponding frequency.
+ """,
+ extended = """
+ Examples:
+ > SELECT ngrams(array("abc", "abc", "bcd", "abc", "bcd"), 2, 4);
+ [{["abc","bcd"]:2.0},
+ {["abc","abc"]:1.0},
+ {["bcd","abc"]:1.0}]
+ """)
+case class NGrams(
+ child: Expression,
+ nExpression: Expression,
+ kExpression: Expression,
+ accuracyExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int)
+ extends TypedImperativeAggregate[NGramBuffer] with ImplicitCastInputTypes {
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression,
+ accuracyExpression: Expression) = {
+ this(child, nExpression, kExpression, NGrams.getAccuracy(kExpression, accuracyExpression), 0, 0)
+ }
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression) = {
+ this(child, nExpression, kExpression, Literal(0))
+ }
+
+ private lazy val n: Int = nExpression.eval().asInstanceOf[Int]
+ private lazy val k: Int = kExpression.eval().asInstanceOf[Int]
+ private lazy val accuracy: Int = accuracyExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(ArrayType(StringType, false), ArrayType(ArrayType(StringType, false))),
--- End diff --
If the array contains null, n-gram of that array will be ambiguous to evaluate. Should we just ignore the null value? For example regard n-gram of `array(\u2018a\u2019, null, 'b')` the same as `array(\u2018a\u2019, 'b')`. Maybe we should refuse the array containing null.
---
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 #17359: [SPARK-20028][SQL] Add aggreagate expression nGrams
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on the issue:
https://github.com/apache/spark/pull/17359
@gatorsmile I will try to take a look again.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #17359: [SPARK-20028][SQL] Add aggreagate expression nGrams
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/17359
Can one of the admins verify this patch?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #17359: [SPARK-20028][SQL] Add aggreagate expression nGra...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/17359#discussion_r107347983
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/NGrams.scala ---
@@ -0,0 +1,258 @@
+/*
+ * 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 java.util.HashMap
+
+import org.apache.spark.serializer.KryoSerializer
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult._
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription, ImplicitCastInputTypes, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.NGrams.NGramBuffer
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData}
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Return the top-k n-grams in rows that consist of sequences of strings.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(expr, n, k, accuracy) - Estimates the top-k n-grams in rows that consist of sequences
+ of strings, represented as arrays of strings, or arrays of arrays of strings. 'accuracy' is an
+ optional precision factor that controls memory usage.
+ The parameter 'n' specifies what type of n-grams are being estimated. Unigrams are n = 1, and
+ bigrams are n = 2. Generally, n will not be greater than about 5. The 'k' parameter specifies
+ how many of the highest-frequency n-grams will be returned by the UDAF. The optional precision
+ factor 'accuracy' specifies how much memory to use for estimation; more memory will give
+ more accurate frequency counts, but could crash the JVM. The value will be the max between
+ 'accuracy'(0 if it's not specified) and 1000/k, which indicates the max number of n-grams
+ which are kept in the internal HashMap.
+ The output is an array of maps with the top-k n-grams and corresponding frequency.
+ """,
+ extended = """
+ Examples:
+ > SELECT ngrams(array("abc", "abc", "bcd", "abc", "bcd"), 2, 4);
+ [{["abc","bcd"]:2.0},
+ {["abc","abc"]:1.0},
+ {["bcd","abc"]:1.0}]
+ """)
+case class NGrams(
+ child: Expression,
+ nExpression: Expression,
+ kExpression: Expression,
+ accuracyExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int)
+ extends TypedImperativeAggregate[NGramBuffer] with ImplicitCastInputTypes {
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression,
+ accuracyExpression: Expression) = {
+ this(child, nExpression, kExpression, NGrams.getAccuracy(kExpression, accuracyExpression), 0, 0)
+ }
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression) = {
+ this(child, nExpression, kExpression, Literal(0))
+ }
+
+ private lazy val n: Int = nExpression.eval().asInstanceOf[Int]
+ private lazy val k: Int = kExpression.eval().asInstanceOf[Int]
+ private lazy val accuracy: Int = accuracyExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(ArrayType(StringType, false), ArrayType(ArrayType(StringType, false))),
+ IntegerType, IntegerType, IntegerType)
+ }
+
+ val isArrayOfString = child.dataType == ArrayType(StringType, false) ||
+ child.dataType == ArrayType(StringType, true)
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (!nExpression.foldable || !kExpression.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 {
+ TypeCheckSuccess
+ }
+ }
+
+ override def createAggregationBuffer(): NGramBuffer = {
+ new NGramBuffer(n, k, accuracy, new HashMap[Vector[UTF8String], Double]())
+ }
+
+ def updateArray(genericArrayData: GenericArrayData, buffer: NGramBuffer, inputRow: InternalRow) {
+ val values = (0 until genericArrayData.numElements()).map(genericArrayData.get(_, StringType).
+ asInstanceOf[UTF8String]).toVector
+ val nGrams = getNGrams(values, n)
+ nGrams.foreach(buffer.add(_))
+ buffer.trim()
+ }
+ override def update(buffer: NGramBuffer, inputRow: InternalRow): NGramBuffer = {
+ if (isArrayOfString) {
+ updateArray(child.eval(inputRow).asInstanceOf[GenericArrayData], buffer, inputRow)
+ }
+ else {
+ val arrayOfArray = child.eval(inputRow).asInstanceOf[GenericArrayData]
+ for (i <- 0 until arrayOfArray.numElements()) {
+ updateArray(arrayOfArray.getArray(i).asInstanceOf[GenericArrayData], buffer, inputRow)
+ }
+ }
+ buffer
+ }
+
+ override def merge(buffer: NGramBuffer, input: NGramBuffer): NGramBuffer = {
+ buffer.merge(input)
+ buffer.trim()
+ buffer
+ }
+
+ override def eval(buffer: NGramBuffer): Any = {
+ val topKNGrams = buffer.getTopKNGrams().map((keyValuePair: (Vector[UTF8String], Double)) => {
+ val arrayKey = new GenericArrayData(Vector(new GenericArrayData(keyValuePair._1)))
+ val arrayValue = new GenericArrayData(Vector(keyValuePair._2))
+ new ArrayBasedMapData(arrayKey, arrayValue)
+ }).toVector
+ new GenericArrayData(topKNGrams)
+ }
+
+ private def getNGrams(values: Vector[UTF8String], n: Int): Vector[Vector[UTF8String]] = {
+ if (values.length >= n) {
+ values.sliding(n).toVector
+ }
+ else {
+ Vector()
+ }
+ }
+
+ override def withNewMutableAggBufferOffset(newOffset: Int): NGrams =
+ copy(mutableAggBufferOffset = newOffset)
+
+ override def withNewInputAggBufferOffset(newOffset: Int): NGrams =
+ copy(inputAggBufferOffset = newOffset)
+
+ override def children: Seq[Expression] = Seq(child, nExpression, kExpression, accuracyExpression)
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = ArrayType(MapType(ArrayType(StringType), DoubleType))
+
+ override def prettyName: String = "ngrams"
+
+ override def serialize(obj: NGramBuffer): Array[Byte] = {
+ NGrams.serializer.serialize(obj)
+ }
+
+ override def deserialize(bytes: Array[Byte]): NGramBuffer = {
+ NGrams.serializer.deserialize(bytes)
+ }
+}
+
+object NGrams {
+
+ val DEFAULT_ACCURACY: Int = 1000
+
+ private def getAccuracy(kExpression: Expression, accuracyExpression: Expression): Expression = {
+ val accuracy = accuracyExpression.eval().asInstanceOf[Int]
+ val k = kExpression.eval().asInstanceOf[Int]
+ Literal(accuracy.max(DEFAULT_ACCURACY / k))
+ }
+
+ val kryoSerializer: KryoSerializer = new KryoSerializer(new SparkConf())
+
+ import collection.JavaConverters._
+
+ class NGramBuffer(val n: Int,
+ val k: Int,
+ val precisionFactor: Int,
+ val frequencyMap: HashMap[Vector[UTF8String], Double]) {
+ def add(ng: Vector[UTF8String]): Unit = {
+ var currentFrequency: Double = frequencyMap.get(ng)
+ if (currentFrequency == null.asInstanceOf[Double]) {
--- End diff --
Check existence by `HashMap`'s API.
---
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 #17359: [SPARK-20028][SQL] Add aggreagate expression nGrams
Posted by sijunhe <gi...@git.apache.org>.
Github user sijunhe commented on the issue:
https://github.com/apache/spark/pull/17359
Would love to see this feature in Spark SQL.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #17359: [SPARK-20028][SQL] Add aggreagate expression nGra...
Posted by gczsjdy <gi...@git.apache.org>.
Github user gczsjdy commented on a diff in the pull request:
https://github.com/apache/spark/pull/17359#discussion_r107444455
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/NGrams.scala ---
@@ -0,0 +1,258 @@
+/*
+ * 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 java.util.HashMap
+
+import org.apache.spark.serializer.KryoSerializer
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult._
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription, ImplicitCastInputTypes, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.NGrams.NGramBuffer
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData}
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Return the top-k n-grams in rows that consist of sequences of strings.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(expr, n, k, accuracy) - Estimates the top-k n-grams in rows that consist of sequences
+ of strings, represented as arrays of strings, or arrays of arrays of strings. 'accuracy' is an
+ optional precision factor that controls memory usage.
+ The parameter 'n' specifies what type of n-grams are being estimated. Unigrams are n = 1, and
+ bigrams are n = 2. Generally, n will not be greater than about 5. The 'k' parameter specifies
+ how many of the highest-frequency n-grams will be returned by the UDAF. The optional precision
+ factor 'accuracy' specifies how much memory to use for estimation; more memory will give
+ more accurate frequency counts, but could crash the JVM. The value will be the max between
+ 'accuracy'(0 if it's not specified) and 1000/k, which indicates the max number of n-grams
+ which are kept in the internal HashMap.
+ The output is an array of maps with the top-k n-grams and corresponding frequency.
+ """,
+ extended = """
+ Examples:
+ > SELECT ngrams(array("abc", "abc", "bcd", "abc", "bcd"), 2, 4);
+ [{["abc","bcd"]:2.0},
+ {["abc","abc"]:1.0},
+ {["bcd","abc"]:1.0}]
+ """)
+case class NGrams(
+ child: Expression,
+ nExpression: Expression,
+ kExpression: Expression,
+ accuracyExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int)
+ extends TypedImperativeAggregate[NGramBuffer] with ImplicitCastInputTypes {
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression,
+ accuracyExpression: Expression) = {
+ this(child, nExpression, kExpression, NGrams.getAccuracy(kExpression, accuracyExpression), 0, 0)
+ }
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression) = {
+ this(child, nExpression, kExpression, Literal(0))
+ }
+
+ private lazy val n: Int = nExpression.eval().asInstanceOf[Int]
+ private lazy val k: Int = kExpression.eval().asInstanceOf[Int]
+ private lazy val accuracy: Int = accuracyExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(ArrayType(StringType, false), ArrayType(ArrayType(StringType, false))),
+ IntegerType, IntegerType, IntegerType)
+ }
+
+ val isArrayOfString = child.dataType == ArrayType(StringType, false) ||
+ child.dataType == ArrayType(StringType, true)
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (!nExpression.foldable || !kExpression.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 {
+ TypeCheckSuccess
+ }
+ }
+
+ override def createAggregationBuffer(): NGramBuffer = {
+ new NGramBuffer(n, k, accuracy, new HashMap[Vector[UTF8String], Double]())
+ }
+
+ def updateArray(genericArrayData: GenericArrayData, buffer: NGramBuffer, inputRow: InternalRow) {
+ val values = (0 until genericArrayData.numElements()).map(genericArrayData.get(_, StringType).
+ asInstanceOf[UTF8String]).toVector
+ val nGrams = getNGrams(values, n)
+ nGrams.foreach(buffer.add(_))
+ buffer.trim()
+ }
+ override def update(buffer: NGramBuffer, inputRow: InternalRow): NGramBuffer = {
+ if (isArrayOfString) {
+ updateArray(child.eval(inputRow).asInstanceOf[GenericArrayData], buffer, inputRow)
+ }
+ else {
+ val arrayOfArray = child.eval(inputRow).asInstanceOf[GenericArrayData]
+ for (i <- 0 until arrayOfArray.numElements()) {
+ updateArray(arrayOfArray.getArray(i).asInstanceOf[GenericArrayData], buffer, inputRow)
+ }
+ }
+ buffer
+ }
+
+ override def merge(buffer: NGramBuffer, input: NGramBuffer): NGramBuffer = {
+ buffer.merge(input)
+ buffer.trim()
+ buffer
+ }
+
+ override def eval(buffer: NGramBuffer): Any = {
+ val topKNGrams = buffer.getTopKNGrams().map((keyValuePair: (Vector[UTF8String], Double)) => {
+ val arrayKey = new GenericArrayData(Vector(new GenericArrayData(keyValuePair._1)))
+ val arrayValue = new GenericArrayData(Vector(keyValuePair._2))
+ new ArrayBasedMapData(arrayKey, arrayValue)
+ }).toVector
+ new GenericArrayData(topKNGrams)
+ }
+
+ private def getNGrams(values: Vector[UTF8String], n: Int): Vector[Vector[UTF8String]] = {
+ if (values.length >= n) {
+ values.sliding(n).toVector
+ }
+ else {
+ Vector()
+ }
+ }
+
+ override def withNewMutableAggBufferOffset(newOffset: Int): NGrams =
+ copy(mutableAggBufferOffset = newOffset)
+
+ override def withNewInputAggBufferOffset(newOffset: Int): NGrams =
+ copy(inputAggBufferOffset = newOffset)
+
+ override def children: Seq[Expression] = Seq(child, nExpression, kExpression, accuracyExpression)
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = ArrayType(MapType(ArrayType(StringType), DoubleType))
+
+ override def prettyName: String = "ngrams"
+
+ override def serialize(obj: NGramBuffer): Array[Byte] = {
+ NGrams.serializer.serialize(obj)
+ }
+
+ override def deserialize(bytes: Array[Byte]): NGramBuffer = {
+ NGrams.serializer.deserialize(bytes)
+ }
+}
+
+object NGrams {
+
+ val DEFAULT_ACCURACY: Int = 1000
+
+ private def getAccuracy(kExpression: Expression, accuracyExpression: Expression): Expression = {
+ val accuracy = accuracyExpression.eval().asInstanceOf[Int]
+ val k = kExpression.eval().asInstanceOf[Int]
+ Literal(accuracy.max(DEFAULT_ACCURACY / k))
+ }
+
+ val kryoSerializer: KryoSerializer = new KryoSerializer(new SparkConf())
+
+ import collection.JavaConverters._
+
+ class NGramBuffer(val n: Int,
+ val k: Int,
+ val precisionFactor: Int,
+ val frequencyMap: HashMap[Vector[UTF8String], Double]) {
+ def add(ng: Vector[UTF8String]): Unit = {
+ var currentFrequency: Double = frequencyMap.get(ng)
+ if (currentFrequency == null.asInstanceOf[Double]) {
+ currentFrequency = 1.0D
+ } else {
+ currentFrequency += 1
+ }
+ frequencyMap.put(ng, currentFrequency)
+ }
+
+ def merge(other: NGramBuffer): Unit = {
+ other.frequencyMap.asScala.foreach((keyValuePair: (Vector[UTF8String], Double)) => {
+ val key = keyValuePair._1
+ val value = keyValuePair._2
+ val originalValue = frequencyMap.getOrDefault(key, 0.0D)
+ frequencyMap.put(key, originalValue + value)
+ })
+ }
+
+ def sortWithTwoFields(frequencyDescend: Boolean)
+ (keyWithFrequency: Tuple2[Vector[UTF8String], Double],
+ keyWithFrequency2: Tuple2[Vector[UTF8String], Double]): Boolean = {
+ if (keyWithFrequency._2 != keyWithFrequency2._2) {
+ (keyWithFrequency._2 < keyWithFrequency2._2) ^ frequencyDescend
+ }
+ else {
+ val keyVector = keyWithFrequency._1
+ val keyVector2 = keyWithFrequency2._1
+ for (i <- 0 until keyVector.length) {
+ if (keyVector(i) != keyVector2(i)) {
+ return (keyVector(i).compare(keyVector2(i))) < 0
+ }
+ }
+ true
+ }
+ }
+
+ def trim(): Unit = {
+ if (frequencyMap.size() > 2 * k * precisionFactor) {
+ val orderedWithIndex = frequencyMap.asScala.iterator.toVector.
+ sortWith(sortWithTwoFields(frequencyDescend = false)).zipWithIndex
+ orderedWithIndex.takeWhile(_._2 < frequencyMap.size() - k * precisionFactor).map(_._1).
+ foreach(keyValuePair => frequencyMap.remove(keyValuePair._1))
+ }
+ }
+
+ def getTopKNGrams(): Seq[(Vector[UTF8String], Double)] = {
+ frequencyMap.asScala.iterator.toVector.sortWith(sortWithTwoFields(frequencyDescend = true)).
+ zipWithIndex.takeWhile(_._2 < k).map(_._1)
+ }
+
+ }
+
+ class NGramBufferSerializer {
+
+ final def serialize(obj: NGramBuffer): Array[Byte] = {
+ NGrams.kryoSerializer.newInstance().serialize[NGramBuffer](obj).array()
--- End diff --
I tried to reuse the instance `val kryoSerializerInstance = new KryoSerializer(new SparkConf()).newInstance()` but failed, still working on this.
---
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 #17359: [SPARK-20028][SQL] Add aggreagate expression nGra...
Posted by gczsjdy <gi...@git.apache.org>.
Github user gczsjdy commented on a diff in the pull request:
https://github.com/apache/spark/pull/17359#discussion_r110823102
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/NGrams.scala ---
@@ -0,0 +1,249 @@
+/*
+ * 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 java.util.HashMap
+
+import org.apache.spark.serializer.KryoSerializer
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult._
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription, ImplicitCastInputTypes, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.NGrams.NGramBuffer
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData}
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Return the top-k n-grams in rows that consist of sequences of strings.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(expr, n, k, accuracy) - Estimates the top-k n-grams in rows that consist of sequences
+ of strings, represented as arrays of strings, or arrays of arrays of strings. 'accuracy' is an
+ optional precision factor that controls memory usage.
+ The parameter 'n' specifies what type of n-grams are being estimated. Unigrams are n = 1, and
+ bigrams are n = 2. Generally, n will not be greater than about 5. The 'k' parameter specifies
+ how many of the highest-frequency n-grams will be returned by the UDAF. The optional precision
+ factor 'accuracy' specifies how much memory to use for estimation; more memory will give
+ more accurate frequency counts, but could crash the JVM. The value will be the max between
+ 'accuracy'(0 if it's not specified) and 1000/k, which indicates the max number of n-grams
+ which are kept in the internal HashMap.
+ The output is an array of maps with the top-k n-grams and corresponding frequency.
+ """,
+ extended = """
+ Examples:
+ > SELECT ngrams(array("abc", "abc", "bcd", "abc", "bcd"), 2, 4);
+ [{["abc","bcd"]:2.0},
+ {["abc","abc"]:1.0},
+ {["bcd","abc"]:1.0}]
+ """)
+case class NGrams(
+ child: Expression,
+ nExpression: Expression,
+ kExpression: Expression,
+ accuracyExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int)
+ extends TypedImperativeAggregate[NGramBuffer] with ImplicitCastInputTypes {
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression,
+ accuracyExpression: Expression) = {
+ this(child, nExpression, kExpression, NGrams.getAccuracy(kExpression, accuracyExpression), 0, 0)
+ }
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression) = {
+ this(child, nExpression, kExpression, Literal(0))
+ }
+
+ private lazy val n: Int = nExpression.eval().asInstanceOf[Int]
+ private lazy val k: Int = kExpression.eval().asInstanceOf[Int]
+ private lazy val accuracy: Int = accuracyExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(ArrayType(StringType, false), ArrayType(ArrayType(StringType, false))),
+ IntegerType, IntegerType, IntegerType)
+ }
+
+ val isArrayOfString = child.dataType == ArrayType(StringType, false) ||
+ child.dataType == ArrayType(StringType, true)
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (!nExpression.foldable || !kExpression.foldable || !accuracyExpression.foldable) {
+ TypeCheckFailure(s"The accuracy or percentage provided must be a foldable integer expression")
+ } else if (n <= 0) {
+ TypeCheckFailure(s"The n provided must be a positive integer (current value = $n)")
+ } else if (k <= 0) {
+ TypeCheckFailure(s"The k provided must be a positive integer (current value = $k)")
+ } else if (accuracy <= 0) {
+ TypeCheckFailure(
+ s"The accuracy provided must be a positive integer (current value = $accuracy)")
+ } else {
+ TypeCheckSuccess
+ }
+ }
+
+ override def createAggregationBuffer(): NGramBuffer = {
+ new NGramBuffer(n, k, accuracy, new HashMap[Vector[UTF8String], Double]())
+ }
+
+ def updateArray(genericArrayData: GenericArrayData, buffer: NGramBuffer, inputRow: InternalRow) {
+ val values = (0 until genericArrayData.numElements()).map(genericArrayData.get(_, StringType).
+ asInstanceOf[UTF8String]).toVector
+ val nGrams = getNGrams(values, n)
+ nGrams.foreach(buffer.add(_))
+ buffer.trim()
+ }
+ override def update(buffer: NGramBuffer, inputRow: InternalRow): NGramBuffer = {
+ if (isArrayOfString) {
+ updateArray(child.eval(inputRow).asInstanceOf[GenericArrayData], buffer, inputRow)
+ } else {
+ val arrayOfArray = child.eval(inputRow).asInstanceOf[GenericArrayData]
+ for (i <- 0 until arrayOfArray.numElements()) {
+ updateArray(arrayOfArray.getArray(i).asInstanceOf[GenericArrayData], buffer, inputRow)
+ }
+ }
+ buffer
+ }
+
+ override def merge(buffer: NGramBuffer, input: NGramBuffer): NGramBuffer = {
+ buffer.merge(input)
+ buffer.trim()
+ buffer
+ }
+
+ override def eval(buffer: NGramBuffer): Any = {
+ val topKNGrams = buffer.getTopKNGrams().map((keyValuePair: (Vector[UTF8String], Double)) => {
+ val arrayKey = new GenericArrayData(Vector(new GenericArrayData(keyValuePair._1)))
+ val arrayValue = new GenericArrayData(Vector(keyValuePair._2))
+ new ArrayBasedMapData(arrayKey, arrayValue)
+ }).toVector
+ new GenericArrayData(topKNGrams)
+ }
+
+ private def getNGrams(values: Vector[UTF8String], n: Int): Vector[Vector[UTF8String]] = {
+ if (values.length >= n) {
+ values.sliding(n).toVector
+ } else {
+ Vector()
+ }
+ }
+
+ override def withNewMutableAggBufferOffset(newOffset: Int): NGrams =
+ copy(mutableAggBufferOffset = newOffset)
+
+ override def withNewInputAggBufferOffset(newOffset: Int): NGrams =
+ copy(inputAggBufferOffset = newOffset)
+
+ override def children: Seq[Expression] = Seq(child, nExpression, kExpression, accuracyExpression)
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = ArrayType(MapType(ArrayType(StringType), DoubleType))
+
+ override def prettyName: String = "ngrams"
+
+ override def serialize(obj: NGramBuffer): Array[Byte] = {
+ NGrams.serializer.serialize(obj)
+ }
+
+ override def deserialize(bytes: Array[Byte]): NGramBuffer = {
+ NGrams.serializer.deserialize(bytes)
+ }
+}
+
+object NGrams {
+
+ val DEFAULT_ACCURACY: Int = 1000
+
+ private def getAccuracy(kExpression: Expression, accuracyExpression: Expression): Expression = {
+ val accuracy = accuracyExpression.eval().asInstanceOf[Int]
+ val k = kExpression.eval().asInstanceOf[Int]
+ Literal(accuracy.max(DEFAULT_ACCURACY / k))
+ }
+
+ val kryoSerializer: KryoSerializer = new KryoSerializer(new SparkConf())
+
+ import collection.JavaConverters._
+
+ class NGramBuffer(val n: Int,
+ val k: Int,
+ val precisionFactor: Int,
+ val frequencyMap: HashMap[Vector[UTF8String], Double]) {
--- End diff --
You are right, it doesn't need to be. By the way, what's the community's major concerns about merging the [expression](https://github.com/apache/spark/pull/16476)/aggregate expression PRs? The performance?
---
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 #17359: [SPARK-20028][SQL] Add aggreagate expression nGra...
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/17359#discussion_r110569002
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/NGrams.scala ---
@@ -0,0 +1,258 @@
+/*
+ * 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 java.util.HashMap
+
+import org.apache.spark.serializer.KryoSerializer
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult._
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription, ImplicitCastInputTypes, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.NGrams.NGramBuffer
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData}
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Return the top-k n-grams in rows that consist of sequences of strings.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(expr, n, k, accuracy) - Estimates the top-k n-grams in rows that consist of sequences
+ of strings, represented as arrays of strings, or arrays of arrays of strings. 'accuracy' is an
+ optional precision factor that controls memory usage.
+ The parameter 'n' specifies what type of n-grams are being estimated. Unigrams are n = 1, and
+ bigrams are n = 2. Generally, n will not be greater than about 5. The 'k' parameter specifies
+ how many of the highest-frequency n-grams will be returned by the UDAF. The optional precision
+ factor 'accuracy' specifies how much memory to use for estimation; more memory will give
+ more accurate frequency counts, but could crash the JVM. The value will be the max between
+ 'accuracy'(0 if it's not specified) and 1000/k, which indicates the max number of n-grams
+ which are kept in the internal HashMap.
+ The output is an array of maps with the top-k n-grams and corresponding frequency.
+ """,
+ extended = """
+ Examples:
+ > SELECT ngrams(array("abc", "abc", "bcd", "abc", "bcd"), 2, 4);
+ [{["abc","bcd"]:2.0},
+ {["abc","abc"]:1.0},
+ {["bcd","abc"]:1.0}]
+ """)
+case class NGrams(
+ child: Expression,
+ nExpression: Expression,
+ kExpression: Expression,
+ accuracyExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int)
+ extends TypedImperativeAggregate[NGramBuffer] with ImplicitCastInputTypes {
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression,
+ accuracyExpression: Expression) = {
+ this(child, nExpression, kExpression, NGrams.getAccuracy(kExpression, accuracyExpression), 0, 0)
+ }
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression) = {
+ this(child, nExpression, kExpression, Literal(0))
+ }
+
+ private lazy val n: Int = nExpression.eval().asInstanceOf[Int]
+ private lazy val k: Int = kExpression.eval().asInstanceOf[Int]
+ private lazy val accuracy: Int = accuracyExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(ArrayType(StringType, false), ArrayType(ArrayType(StringType, false))),
+ IntegerType, IntegerType, IntegerType)
+ }
+
+ val isArrayOfString = child.dataType == ArrayType(StringType, false) ||
+ child.dataType == ArrayType(StringType, true)
--- End diff --
To be honest, I don't there is a way to guarantee no null values at plan compile time.
---
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 #17359: [SPARK-20028][SQL] Add aggreagate expression nGra...
Posted by gczsjdy <gi...@git.apache.org>.
Github user gczsjdy commented on a diff in the pull request:
https://github.com/apache/spark/pull/17359#discussion_r107429714
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/NGrams.scala ---
@@ -0,0 +1,258 @@
+/*
+ * 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 java.util.HashMap
+
+import org.apache.spark.serializer.KryoSerializer
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult._
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription, ImplicitCastInputTypes, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.NGrams.NGramBuffer
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData}
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Return the top-k n-grams in rows that consist of sequences of strings.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(expr, n, k, accuracy) - Estimates the top-k n-grams in rows that consist of sequences
+ of strings, represented as arrays of strings, or arrays of arrays of strings. 'accuracy' is an
+ optional precision factor that controls memory usage.
+ The parameter 'n' specifies what type of n-grams are being estimated. Unigrams are n = 1, and
+ bigrams are n = 2. Generally, n will not be greater than about 5. The 'k' parameter specifies
+ how many of the highest-frequency n-grams will be returned by the UDAF. The optional precision
+ factor 'accuracy' specifies how much memory to use for estimation; more memory will give
+ more accurate frequency counts, but could crash the JVM. The value will be the max between
+ 'accuracy'(0 if it's not specified) and 1000/k, which indicates the max number of n-grams
+ which are kept in the internal HashMap.
+ The output is an array of maps with the top-k n-grams and corresponding frequency.
+ """,
+ extended = """
+ Examples:
+ > SELECT ngrams(array("abc", "abc", "bcd", "abc", "bcd"), 2, 4);
+ [{["abc","bcd"]:2.0},
+ {["abc","abc"]:1.0},
+ {["bcd","abc"]:1.0}]
+ """)
+case class NGrams(
+ child: Expression,
+ nExpression: Expression,
+ kExpression: Expression,
+ accuracyExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int)
+ extends TypedImperativeAggregate[NGramBuffer] with ImplicitCastInputTypes {
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression,
+ accuracyExpression: Expression) = {
+ this(child, nExpression, kExpression, NGrams.getAccuracy(kExpression, accuracyExpression), 0, 0)
+ }
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression) = {
+ this(child, nExpression, kExpression, Literal(0))
+ }
+
+ private lazy val n: Int = nExpression.eval().asInstanceOf[Int]
+ private lazy val k: Int = kExpression.eval().asInstanceOf[Int]
+ private lazy val accuracy: Int = accuracyExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(ArrayType(StringType, false), ArrayType(ArrayType(StringType, false))),
+ IntegerType, IntegerType, IntegerType)
+ }
+
+ val isArrayOfString = child.dataType == ArrayType(StringType, false) ||
+ child.dataType == ArrayType(StringType, true)
--- End diff --
Yes, actually I want to refuse users' input with Array of String with null so I didn't include it in `inputTypes`, but the tests I have done showed me that some Array without null will be of type `ArrayType(StringType, true)`.
---
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 #17359: [SPARK-20028][SQL] Add aggreagate expression nGrams
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on the issue:
https://github.com/apache/spark/pull/17359
Regarding the performance issue, does this change have significant improvement compared with Hive's?
---
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 #17359: [SPARK-20028][SQL] Add aggreagate expression nGra...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/17359#discussion_r107350167
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/NGrams.scala ---
@@ -0,0 +1,258 @@
+/*
+ * 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 java.util.HashMap
+
+import org.apache.spark.serializer.KryoSerializer
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult._
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription, ImplicitCastInputTypes, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.NGrams.NGramBuffer
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData}
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Return the top-k n-grams in rows that consist of sequences of strings.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(expr, n, k, accuracy) - Estimates the top-k n-grams in rows that consist of sequences
+ of strings, represented as arrays of strings, or arrays of arrays of strings. 'accuracy' is an
+ optional precision factor that controls memory usage.
+ The parameter 'n' specifies what type of n-grams are being estimated. Unigrams are n = 1, and
+ bigrams are n = 2. Generally, n will not be greater than about 5. The 'k' parameter specifies
+ how many of the highest-frequency n-grams will be returned by the UDAF. The optional precision
+ factor 'accuracy' specifies how much memory to use for estimation; more memory will give
+ more accurate frequency counts, but could crash the JVM. The value will be the max between
+ 'accuracy'(0 if it's not specified) and 1000/k, which indicates the max number of n-grams
+ which are kept in the internal HashMap.
+ The output is an array of maps with the top-k n-grams and corresponding frequency.
+ """,
+ extended = """
+ Examples:
+ > SELECT ngrams(array("abc", "abc", "bcd", "abc", "bcd"), 2, 4);
+ [{["abc","bcd"]:2.0},
+ {["abc","abc"]:1.0},
+ {["bcd","abc"]:1.0}]
+ """)
+case class NGrams(
+ child: Expression,
+ nExpression: Expression,
+ kExpression: Expression,
+ accuracyExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int)
+ extends TypedImperativeAggregate[NGramBuffer] with ImplicitCastInputTypes {
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression,
+ accuracyExpression: Expression) = {
+ this(child, nExpression, kExpression, NGrams.getAccuracy(kExpression, accuracyExpression), 0, 0)
+ }
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression) = {
+ this(child, nExpression, kExpression, Literal(0))
+ }
+
+ private lazy val n: Int = nExpression.eval().asInstanceOf[Int]
+ private lazy val k: Int = kExpression.eval().asInstanceOf[Int]
+ private lazy val accuracy: Int = accuracyExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(ArrayType(StringType, false), ArrayType(ArrayType(StringType, false))),
+ IntegerType, IntegerType, IntegerType)
+ }
+
+ val isArrayOfString = child.dataType == ArrayType(StringType, false) ||
+ child.dataType == ArrayType(StringType, true)
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (!nExpression.foldable || !kExpression.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 {
+ TypeCheckSuccess
+ }
+ }
+
+ override def createAggregationBuffer(): NGramBuffer = {
+ new NGramBuffer(n, k, accuracy, new HashMap[Vector[UTF8String], Double]())
+ }
+
+ def updateArray(genericArrayData: GenericArrayData, buffer: NGramBuffer, inputRow: InternalRow) {
+ val values = (0 until genericArrayData.numElements()).map(genericArrayData.get(_, StringType).
+ asInstanceOf[UTF8String]).toVector
+ val nGrams = getNGrams(values, n)
+ nGrams.foreach(buffer.add(_))
+ buffer.trim()
+ }
+ override def update(buffer: NGramBuffer, inputRow: InternalRow): NGramBuffer = {
+ if (isArrayOfString) {
+ updateArray(child.eval(inputRow).asInstanceOf[GenericArrayData], buffer, inputRow)
+ }
+ else {
+ val arrayOfArray = child.eval(inputRow).asInstanceOf[GenericArrayData]
+ for (i <- 0 until arrayOfArray.numElements()) {
+ updateArray(arrayOfArray.getArray(i).asInstanceOf[GenericArrayData], buffer, inputRow)
+ }
+ }
+ buffer
+ }
+
+ override def merge(buffer: NGramBuffer, input: NGramBuffer): NGramBuffer = {
+ buffer.merge(input)
+ buffer.trim()
+ buffer
+ }
+
+ override def eval(buffer: NGramBuffer): Any = {
+ val topKNGrams = buffer.getTopKNGrams().map((keyValuePair: (Vector[UTF8String], Double)) => {
+ val arrayKey = new GenericArrayData(Vector(new GenericArrayData(keyValuePair._1)))
+ val arrayValue = new GenericArrayData(Vector(keyValuePair._2))
+ new ArrayBasedMapData(arrayKey, arrayValue)
+ }).toVector
+ new GenericArrayData(topKNGrams)
+ }
+
+ private def getNGrams(values: Vector[UTF8String], n: Int): Vector[Vector[UTF8String]] = {
+ if (values.length >= n) {
+ values.sliding(n).toVector
+ }
+ else {
+ Vector()
+ }
+ }
+
+ override def withNewMutableAggBufferOffset(newOffset: Int): NGrams =
+ copy(mutableAggBufferOffset = newOffset)
+
+ override def withNewInputAggBufferOffset(newOffset: Int): NGrams =
+ copy(inputAggBufferOffset = newOffset)
+
+ override def children: Seq[Expression] = Seq(child, nExpression, kExpression, accuracyExpression)
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = ArrayType(MapType(ArrayType(StringType), DoubleType))
+
+ override def prettyName: String = "ngrams"
+
+ override def serialize(obj: NGramBuffer): Array[Byte] = {
+ NGrams.serializer.serialize(obj)
+ }
+
+ override def deserialize(bytes: Array[Byte]): NGramBuffer = {
+ NGrams.serializer.deserialize(bytes)
+ }
+}
+
+object NGrams {
+
+ val DEFAULT_ACCURACY: Int = 1000
+
+ private def getAccuracy(kExpression: Expression, accuracyExpression: Expression): Expression = {
+ val accuracy = accuracyExpression.eval().asInstanceOf[Int]
+ val k = kExpression.eval().asInstanceOf[Int]
+ Literal(accuracy.max(DEFAULT_ACCURACY / k))
+ }
+
+ val kryoSerializer: KryoSerializer = new KryoSerializer(new SparkConf())
+
+ import collection.JavaConverters._
+
+ class NGramBuffer(val n: Int,
+ val k: Int,
+ val precisionFactor: Int,
+ val frequencyMap: HashMap[Vector[UTF8String], Double]) {
+ def add(ng: Vector[UTF8String]): Unit = {
+ var currentFrequency: Double = frequencyMap.get(ng)
+ if (currentFrequency == null.asInstanceOf[Double]) {
+ currentFrequency = 1.0D
+ } else {
+ currentFrequency += 1
+ }
+ frequencyMap.put(ng, currentFrequency)
+ }
+
+ def merge(other: NGramBuffer): Unit = {
+ other.frequencyMap.asScala.foreach((keyValuePair: (Vector[UTF8String], Double)) => {
+ val key = keyValuePair._1
+ val value = keyValuePair._2
+ val originalValue = frequencyMap.getOrDefault(key, 0.0D)
+ frequencyMap.put(key, originalValue + value)
+ })
+ }
+
+ def sortWithTwoFields(frequencyDescend: Boolean)
+ (keyWithFrequency: Tuple2[Vector[UTF8String], Double],
+ keyWithFrequency2: Tuple2[Vector[UTF8String], Double]): Boolean = {
+ if (keyWithFrequency._2 != keyWithFrequency2._2) {
+ (keyWithFrequency._2 < keyWithFrequency2._2) ^ frequencyDescend
+ }
+ else {
--- End diff --
style. `} else {`
---
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 #17359: [SPARK-20028][SQL] Add aggreagate expression nGrams
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/17359
Can one of the admins verify this patch?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #17359: [SPARK-20028][SQL] Add aggreagate expression nGra...
Posted by gczsjdy <gi...@git.apache.org>.
Github user gczsjdy commented on a diff in the pull request:
https://github.com/apache/spark/pull/17359#discussion_r107417310
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/NGrams.scala ---
@@ -0,0 +1,258 @@
+/*
+ * 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 java.util.HashMap
+
+import org.apache.spark.serializer.KryoSerializer
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult._
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription, ImplicitCastInputTypes, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.NGrams.NGramBuffer
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData}
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Return the top-k n-grams in rows that consist of sequences of strings.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(expr, n, k, accuracy) - Estimates the top-k n-grams in rows that consist of sequences
+ of strings, represented as arrays of strings, or arrays of arrays of strings. 'accuracy' is an
+ optional precision factor that controls memory usage.
+ The parameter 'n' specifies what type of n-grams are being estimated. Unigrams are n = 1, and
+ bigrams are n = 2. Generally, n will not be greater than about 5. The 'k' parameter specifies
+ how many of the highest-frequency n-grams will be returned by the UDAF. The optional precision
+ factor 'accuracy' specifies how much memory to use for estimation; more memory will give
+ more accurate frequency counts, but could crash the JVM. The value will be the max between
+ 'accuracy'(0 if it's not specified) and 1000/k, which indicates the max number of n-grams
+ which are kept in the internal HashMap.
+ The output is an array of maps with the top-k n-grams and corresponding frequency.
+ """,
+ extended = """
+ Examples:
+ > SELECT ngrams(array("abc", "abc", "bcd", "abc", "bcd"), 2, 4);
+ [{["abc","bcd"]:2.0},
+ {["abc","abc"]:1.0},
+ {["bcd","abc"]:1.0}]
+ """)
+case class NGrams(
+ child: Expression,
+ nExpression: Expression,
+ kExpression: Expression,
+ accuracyExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int)
+ extends TypedImperativeAggregate[NGramBuffer] with ImplicitCastInputTypes {
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression,
+ accuracyExpression: Expression) = {
+ this(child, nExpression, kExpression, NGrams.getAccuracy(kExpression, accuracyExpression), 0, 0)
+ }
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression) = {
+ this(child, nExpression, kExpression, Literal(0))
+ }
+
+ private lazy val n: Int = nExpression.eval().asInstanceOf[Int]
+ private lazy val k: Int = kExpression.eval().asInstanceOf[Int]
+ private lazy val accuracy: Int = accuracyExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(ArrayType(StringType, false), ArrayType(ArrayType(StringType, false))),
+ IntegerType, IntegerType, IntegerType)
+ }
+
+ val isArrayOfString = child.dataType == ArrayType(StringType, false) ||
+ child.dataType == ArrayType(StringType, true)
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (!nExpression.foldable || !kExpression.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 {
+ TypeCheckSuccess
+ }
+ }
+
+ override def createAggregationBuffer(): NGramBuffer = {
+ new NGramBuffer(n, k, accuracy, new HashMap[Vector[UTF8String], Double]())
+ }
+
+ def updateArray(genericArrayData: GenericArrayData, buffer: NGramBuffer, inputRow: InternalRow) {
+ val values = (0 until genericArrayData.numElements()).map(genericArrayData.get(_, StringType).
+ asInstanceOf[UTF8String]).toVector
+ val nGrams = getNGrams(values, n)
+ nGrams.foreach(buffer.add(_))
+ buffer.trim()
+ }
+ override def update(buffer: NGramBuffer, inputRow: InternalRow): NGramBuffer = {
+ if (isArrayOfString) {
+ updateArray(child.eval(inputRow).asInstanceOf[GenericArrayData], buffer, inputRow)
+ }
+ else {
+ val arrayOfArray = child.eval(inputRow).asInstanceOf[GenericArrayData]
+ for (i <- 0 until arrayOfArray.numElements()) {
+ updateArray(arrayOfArray.getArray(i).asInstanceOf[GenericArrayData], buffer, inputRow)
+ }
+ }
+ buffer
+ }
+
+ override def merge(buffer: NGramBuffer, input: NGramBuffer): NGramBuffer = {
+ buffer.merge(input)
+ buffer.trim()
+ buffer
+ }
+
+ override def eval(buffer: NGramBuffer): Any = {
+ val topKNGrams = buffer.getTopKNGrams().map((keyValuePair: (Vector[UTF8String], Double)) => {
+ val arrayKey = new GenericArrayData(Vector(new GenericArrayData(keyValuePair._1)))
+ val arrayValue = new GenericArrayData(Vector(keyValuePair._2))
+ new ArrayBasedMapData(arrayKey, arrayValue)
+ }).toVector
+ new GenericArrayData(topKNGrams)
+ }
+
+ private def getNGrams(values: Vector[UTF8String], n: Int): Vector[Vector[UTF8String]] = {
+ if (values.length >= n) {
+ values.sliding(n).toVector
+ }
+ else {
+ Vector()
+ }
+ }
+
+ override def withNewMutableAggBufferOffset(newOffset: Int): NGrams =
+ copy(mutableAggBufferOffset = newOffset)
+
+ override def withNewInputAggBufferOffset(newOffset: Int): NGrams =
+ copy(inputAggBufferOffset = newOffset)
+
+ override def children: Seq[Expression] = Seq(child, nExpression, kExpression, accuracyExpression)
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = ArrayType(MapType(ArrayType(StringType), DoubleType))
+
+ override def prettyName: String = "ngrams"
+
+ override def serialize(obj: NGramBuffer): Array[Byte] = {
+ NGrams.serializer.serialize(obj)
+ }
+
+ override def deserialize(bytes: Array[Byte]): NGramBuffer = {
+ NGrams.serializer.deserialize(bytes)
+ }
+}
+
+object NGrams {
+
+ val DEFAULT_ACCURACY: Int = 1000
+
+ private def getAccuracy(kExpression: Expression, accuracyExpression: Expression): Expression = {
+ val accuracy = accuracyExpression.eval().asInstanceOf[Int]
+ val k = kExpression.eval().asInstanceOf[Int]
+ Literal(accuracy.max(DEFAULT_ACCURACY / k))
+ }
+
+ val kryoSerializer: KryoSerializer = new KryoSerializer(new SparkConf())
+
+ import collection.JavaConverters._
+
+ class NGramBuffer(val n: Int,
+ val k: Int,
+ val precisionFactor: Int,
+ val frequencyMap: HashMap[Vector[UTF8String], Double]) {
+ def add(ng: Vector[UTF8String]): Unit = {
+ var currentFrequency: Double = frequencyMap.get(ng)
+ if (currentFrequency == null.asInstanceOf[Double]) {
--- End diff --
Good way!
---
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 #17359: [SPARK-20028][SQL] Add aggreagate expression nGra...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/17359#discussion_r107343519
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/NGrams.scala ---
@@ -0,0 +1,258 @@
+/*
+ * 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 java.util.HashMap
+
+import org.apache.spark.serializer.KryoSerializer
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult._
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription, ImplicitCastInputTypes, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.NGrams.NGramBuffer
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData}
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Return the top-k n-grams in rows that consist of sequences of strings.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(expr, n, k, accuracy) - Estimates the top-k n-grams in rows that consist of sequences
+ of strings, represented as arrays of strings, or arrays of arrays of strings. 'accuracy' is an
+ optional precision factor that controls memory usage.
+ The parameter 'n' specifies what type of n-grams are being estimated. Unigrams are n = 1, and
+ bigrams are n = 2. Generally, n will not be greater than about 5. The 'k' parameter specifies
+ how many of the highest-frequency n-grams will be returned by the UDAF. The optional precision
+ factor 'accuracy' specifies how much memory to use for estimation; more memory will give
+ more accurate frequency counts, but could crash the JVM. The value will be the max between
+ 'accuracy'(0 if it's not specified) and 1000/k, which indicates the max number of n-grams
+ which are kept in the internal HashMap.
+ The output is an array of maps with the top-k n-grams and corresponding frequency.
+ """,
+ extended = """
+ Examples:
+ > SELECT ngrams(array('abc', 'abc', 'bcd', 'abc', 'bcd'), 2, 4);
+ [["abc","bcd"]:2.0},
--- End diff --
hmm, doesn't "abc" show 3 times?
---
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 #17359: [SPARK-20028][SQL] Add aggreagate expression nGra...
Posted by gczsjdy <gi...@git.apache.org>.
Github user gczsjdy commented on a diff in the pull request:
https://github.com/apache/spark/pull/17359#discussion_r107345616
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/NGrams.scala ---
@@ -0,0 +1,258 @@
+/*
+ * 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 java.util.HashMap
+
+import org.apache.spark.serializer.KryoSerializer
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult._
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription, ImplicitCastInputTypes, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.NGrams.NGramBuffer
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData}
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Return the top-k n-grams in rows that consist of sequences of strings.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(expr, n, k, accuracy) - Estimates the top-k n-grams in rows that consist of sequences
+ of strings, represented as arrays of strings, or arrays of arrays of strings. 'accuracy' is an
+ optional precision factor that controls memory usage.
+ The parameter 'n' specifies what type of n-grams are being estimated. Unigrams are n = 1, and
+ bigrams are n = 2. Generally, n will not be greater than about 5. The 'k' parameter specifies
+ how many of the highest-frequency n-grams will be returned by the UDAF. The optional precision
+ factor 'accuracy' specifies how much memory to use for estimation; more memory will give
+ more accurate frequency counts, but could crash the JVM. The value will be the max between
+ 'accuracy'(0 if it's not specified) and 1000/k, which indicates the max number of n-grams
+ which are kept in the internal HashMap.
+ The output is an array of maps with the top-k n-grams and corresponding frequency.
+ """,
+ extended = """
+ Examples:
+ > SELECT ngrams(array('abc', 'abc', 'bcd', 'abc', 'bcd'), 2, 4);
+ [["abc","bcd"]:2.0},
--- End diff --
Yeah, and actually it's 'bigram' because n equals to 2 here.
---
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 #17359: [SPARK-20028][SQL] Add aggreagate expression nGra...
Posted by gczsjdy <gi...@git.apache.org>.
Github user gczsjdy commented on a diff in the pull request:
https://github.com/apache/spark/pull/17359#discussion_r107418753
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/NGrams.scala ---
@@ -0,0 +1,258 @@
+/*
+ * 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 java.util.HashMap
+
+import org.apache.spark.serializer.KryoSerializer
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult._
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription, ImplicitCastInputTypes, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.NGrams.NGramBuffer
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData}
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Return the top-k n-grams in rows that consist of sequences of strings.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(expr, n, k, accuracy) - Estimates the top-k n-grams in rows that consist of sequences
+ of strings, represented as arrays of strings, or arrays of arrays of strings. 'accuracy' is an
+ optional precision factor that controls memory usage.
+ The parameter 'n' specifies what type of n-grams are being estimated. Unigrams are n = 1, and
+ bigrams are n = 2. Generally, n will not be greater than about 5. The 'k' parameter specifies
+ how many of the highest-frequency n-grams will be returned by the UDAF. The optional precision
+ factor 'accuracy' specifies how much memory to use for estimation; more memory will give
+ more accurate frequency counts, but could crash the JVM. The value will be the max between
+ 'accuracy'(0 if it's not specified) and 1000/k, which indicates the max number of n-grams
+ which are kept in the internal HashMap.
+ The output is an array of maps with the top-k n-grams and corresponding frequency.
+ """,
+ extended = """
+ Examples:
+ > SELECT ngrams(array("abc", "abc", "bcd", "abc", "bcd"), 2, 4);
+ [{["abc","bcd"]:2.0},
+ {["abc","abc"]:1.0},
+ {["bcd","abc"]:1.0}]
+ """)
+case class NGrams(
+ child: Expression,
+ nExpression: Expression,
+ kExpression: Expression,
+ accuracyExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int)
+ extends TypedImperativeAggregate[NGramBuffer] with ImplicitCastInputTypes {
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression,
+ accuracyExpression: Expression) = {
+ this(child, nExpression, kExpression, NGrams.getAccuracy(kExpression, accuracyExpression), 0, 0)
+ }
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression) = {
+ this(child, nExpression, kExpression, Literal(0))
+ }
+
+ private lazy val n: Int = nExpression.eval().asInstanceOf[Int]
+ private lazy val k: Int = kExpression.eval().asInstanceOf[Int]
+ private lazy val accuracy: Int = accuracyExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(ArrayType(StringType, false), ArrayType(ArrayType(StringType, false))),
+ IntegerType, IntegerType, IntegerType)
+ }
+
+ val isArrayOfString = child.dataType == ArrayType(StringType, false) ||
+ child.dataType == ArrayType(StringType, true)
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (!nExpression.foldable || !kExpression.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 {
+ TypeCheckSuccess
+ }
+ }
+
+ override def createAggregationBuffer(): NGramBuffer = {
+ new NGramBuffer(n, k, accuracy, new HashMap[Vector[UTF8String], Double]())
+ }
+
+ def updateArray(genericArrayData: GenericArrayData, buffer: NGramBuffer, inputRow: InternalRow) {
+ val values = (0 until genericArrayData.numElements()).map(genericArrayData.get(_, StringType).
+ asInstanceOf[UTF8String]).toVector
+ val nGrams = getNGrams(values, n)
+ nGrams.foreach(buffer.add(_))
+ buffer.trim()
+ }
+ override def update(buffer: NGramBuffer, inputRow: InternalRow): NGramBuffer = {
+ if (isArrayOfString) {
+ updateArray(child.eval(inputRow).asInstanceOf[GenericArrayData], buffer, inputRow)
+ }
+ else {
+ val arrayOfArray = child.eval(inputRow).asInstanceOf[GenericArrayData]
+ for (i <- 0 until arrayOfArray.numElements()) {
+ updateArray(arrayOfArray.getArray(i).asInstanceOf[GenericArrayData], buffer, inputRow)
+ }
+ }
+ buffer
+ }
+
+ override def merge(buffer: NGramBuffer, input: NGramBuffer): NGramBuffer = {
+ buffer.merge(input)
+ buffer.trim()
+ buffer
+ }
+
+ override def eval(buffer: NGramBuffer): Any = {
+ val topKNGrams = buffer.getTopKNGrams().map((keyValuePair: (Vector[UTF8String], Double)) => {
+ val arrayKey = new GenericArrayData(Vector(new GenericArrayData(keyValuePair._1)))
+ val arrayValue = new GenericArrayData(Vector(keyValuePair._2))
+ new ArrayBasedMapData(arrayKey, arrayValue)
+ }).toVector
+ new GenericArrayData(topKNGrams)
+ }
+
+ private def getNGrams(values: Vector[UTF8String], n: Int): Vector[Vector[UTF8String]] = {
+ if (values.length >= n) {
+ values.sliding(n).toVector
+ }
+ else {
+ Vector()
+ }
+ }
+
+ override def withNewMutableAggBufferOffset(newOffset: Int): NGrams =
+ copy(mutableAggBufferOffset = newOffset)
+
+ override def withNewInputAggBufferOffset(newOffset: Int): NGrams =
+ copy(inputAggBufferOffset = newOffset)
+
+ override def children: Seq[Expression] = Seq(child, nExpression, kExpression, accuracyExpression)
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = ArrayType(MapType(ArrayType(StringType), DoubleType))
+
+ override def prettyName: String = "ngrams"
+
+ override def serialize(obj: NGramBuffer): Array[Byte] = {
+ NGrams.serializer.serialize(obj)
+ }
+
+ override def deserialize(bytes: Array[Byte]): NGramBuffer = {
+ NGrams.serializer.deserialize(bytes)
+ }
+}
+
+object NGrams {
+
+ val DEFAULT_ACCURACY: Int = 1000
+
+ private def getAccuracy(kExpression: Expression, accuracyExpression: Expression): Expression = {
+ val accuracy = accuracyExpression.eval().asInstanceOf[Int]
+ val k = kExpression.eval().asInstanceOf[Int]
+ Literal(accuracy.max(DEFAULT_ACCURACY / k))
+ }
+
+ val kryoSerializer: KryoSerializer = new KryoSerializer(new SparkConf())
+
+ import collection.JavaConverters._
+
+ class NGramBuffer(val n: Int,
+ val k: Int,
+ val precisionFactor: Int,
+ val frequencyMap: HashMap[Vector[UTF8String], Double]) {
+ def add(ng: Vector[UTF8String]): Unit = {
+ var currentFrequency: Double = frequencyMap.get(ng)
+ if (currentFrequency == null.asInstanceOf[Double]) {
+ currentFrequency = 1.0D
+ } else {
+ currentFrequency += 1
+ }
+ frequencyMap.put(ng, currentFrequency)
+ }
+
+ def merge(other: NGramBuffer): Unit = {
+ other.frequencyMap.asScala.foreach((keyValuePair: (Vector[UTF8String], Double)) => {
+ val key = keyValuePair._1
+ val value = keyValuePair._2
+ val originalValue = frequencyMap.getOrDefault(key, 0.0D)
+ frequencyMap.put(key, originalValue + value)
--- End diff --
The doc says `Note that support for Java 7 is deprecated as of Spark 2.0.0 and may be removed in Spark 2.2.0.`, maybe I changed the code first and wait for others to comment.
---
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 #17359: [SPARK-20028][SQL] Add aggreagate expression nGrams
Posted by gczsjdy <gi...@git.apache.org>.
Github user gczsjdy commented on the issue:
https://github.com/apache/spark/pull/17359
cc @chenghao-intel @yucai @adrian-wang @cloud-fan @gatorsmile
---
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 #17359: [SPARK-20028][SQL] Add aggreagate expression nGrams
Posted by gczsjdy <gi...@git.apache.org>.
Github user gczsjdy commented on the issue:
https://github.com/apache/spark/pull/17359
@viirya Updated, as the last sentence mentioned, I have tried to make Spark support `GenericUDAFResolver`. But it lacks some interfaces comparing with `AbstractGenericUDAFResolver` so this can't be done.
---
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 #17359: [SPARK-20028][SQL] Add aggreagate expression nGra...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/17359#discussion_r107352740
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/NGrams.scala ---
@@ -0,0 +1,258 @@
+/*
+ * 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 java.util.HashMap
+
+import org.apache.spark.serializer.KryoSerializer
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult._
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription, ImplicitCastInputTypes, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.NGrams.NGramBuffer
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData}
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Return the top-k n-grams in rows that consist of sequences of strings.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(expr, n, k, accuracy) - Estimates the top-k n-grams in rows that consist of sequences
+ of strings, represented as arrays of strings, or arrays of arrays of strings. 'accuracy' is an
+ optional precision factor that controls memory usage.
+ The parameter 'n' specifies what type of n-grams are being estimated. Unigrams are n = 1, and
+ bigrams are n = 2. Generally, n will not be greater than about 5. The 'k' parameter specifies
+ how many of the highest-frequency n-grams will be returned by the UDAF. The optional precision
+ factor 'accuracy' specifies how much memory to use for estimation; more memory will give
+ more accurate frequency counts, but could crash the JVM. The value will be the max between
+ 'accuracy'(0 if it's not specified) and 1000/k, which indicates the max number of n-grams
+ which are kept in the internal HashMap.
+ The output is an array of maps with the top-k n-grams and corresponding frequency.
+ """,
+ extended = """
+ Examples:
+ > SELECT ngrams(array("abc", "abc", "bcd", "abc", "bcd"), 2, 4);
+ [{["abc","bcd"]:2.0},
+ {["abc","abc"]:1.0},
+ {["bcd","abc"]:1.0}]
+ """)
+case class NGrams(
+ child: Expression,
+ nExpression: Expression,
+ kExpression: Expression,
+ accuracyExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int)
+ extends TypedImperativeAggregate[NGramBuffer] with ImplicitCastInputTypes {
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression,
+ accuracyExpression: Expression) = {
+ this(child, nExpression, kExpression, NGrams.getAccuracy(kExpression, accuracyExpression), 0, 0)
+ }
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression) = {
+ this(child, nExpression, kExpression, Literal(0))
+ }
+
+ private lazy val n: Int = nExpression.eval().asInstanceOf[Int]
+ private lazy val k: Int = kExpression.eval().asInstanceOf[Int]
+ private lazy val accuracy: Int = accuracyExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(ArrayType(StringType, false), ArrayType(ArrayType(StringType, false))),
+ IntegerType, IntegerType, IntegerType)
+ }
+
+ val isArrayOfString = child.dataType == ArrayType(StringType, false) ||
+ child.dataType == ArrayType(StringType, true)
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (!nExpression.foldable || !kExpression.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 {
+ TypeCheckSuccess
+ }
+ }
+
+ override def createAggregationBuffer(): NGramBuffer = {
+ new NGramBuffer(n, k, accuracy, new HashMap[Vector[UTF8String], Double]())
+ }
+
+ def updateArray(genericArrayData: GenericArrayData, buffer: NGramBuffer, inputRow: InternalRow) {
+ val values = (0 until genericArrayData.numElements()).map(genericArrayData.get(_, StringType).
+ asInstanceOf[UTF8String]).toVector
+ val nGrams = getNGrams(values, n)
+ nGrams.foreach(buffer.add(_))
+ buffer.trim()
+ }
+ override def update(buffer: NGramBuffer, inputRow: InternalRow): NGramBuffer = {
+ if (isArrayOfString) {
+ updateArray(child.eval(inputRow).asInstanceOf[GenericArrayData], buffer, inputRow)
+ }
+ else {
+ val arrayOfArray = child.eval(inputRow).asInstanceOf[GenericArrayData]
+ for (i <- 0 until arrayOfArray.numElements()) {
+ updateArray(arrayOfArray.getArray(i).asInstanceOf[GenericArrayData], buffer, inputRow)
+ }
+ }
+ buffer
+ }
+
+ override def merge(buffer: NGramBuffer, input: NGramBuffer): NGramBuffer = {
+ buffer.merge(input)
+ buffer.trim()
+ buffer
+ }
+
+ override def eval(buffer: NGramBuffer): Any = {
+ val topKNGrams = buffer.getTopKNGrams().map((keyValuePair: (Vector[UTF8String], Double)) => {
+ val arrayKey = new GenericArrayData(Vector(new GenericArrayData(keyValuePair._1)))
+ val arrayValue = new GenericArrayData(Vector(keyValuePair._2))
+ new ArrayBasedMapData(arrayKey, arrayValue)
+ }).toVector
+ new GenericArrayData(topKNGrams)
+ }
+
+ private def getNGrams(values: Vector[UTF8String], n: Int): Vector[Vector[UTF8String]] = {
+ if (values.length >= n) {
+ values.sliding(n).toVector
+ }
+ else {
+ Vector()
+ }
+ }
+
+ override def withNewMutableAggBufferOffset(newOffset: Int): NGrams =
+ copy(mutableAggBufferOffset = newOffset)
+
+ override def withNewInputAggBufferOffset(newOffset: Int): NGrams =
+ copy(inputAggBufferOffset = newOffset)
+
+ override def children: Seq[Expression] = Seq(child, nExpression, kExpression, accuracyExpression)
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = ArrayType(MapType(ArrayType(StringType), DoubleType))
+
+ override def prettyName: String = "ngrams"
+
+ override def serialize(obj: NGramBuffer): Array[Byte] = {
+ NGrams.serializer.serialize(obj)
+ }
+
+ override def deserialize(bytes: Array[Byte]): NGramBuffer = {
+ NGrams.serializer.deserialize(bytes)
+ }
+}
+
+object NGrams {
+
+ val DEFAULT_ACCURACY: Int = 1000
+
+ private def getAccuracy(kExpression: Expression, accuracyExpression: Expression): Expression = {
+ val accuracy = accuracyExpression.eval().asInstanceOf[Int]
+ val k = kExpression.eval().asInstanceOf[Int]
+ Literal(accuracy.max(DEFAULT_ACCURACY / k))
+ }
+
+ val kryoSerializer: KryoSerializer = new KryoSerializer(new SparkConf())
+
+ import collection.JavaConverters._
+
+ class NGramBuffer(val n: Int,
+ val k: Int,
+ val precisionFactor: Int,
+ val frequencyMap: HashMap[Vector[UTF8String], Double]) {
+ def add(ng: Vector[UTF8String]): Unit = {
+ var currentFrequency: Double = frequencyMap.get(ng)
+ if (currentFrequency == null.asInstanceOf[Double]) {
+ currentFrequency = 1.0D
+ } else {
+ currentFrequency += 1
+ }
+ frequencyMap.put(ng, currentFrequency)
+ }
+
+ def merge(other: NGramBuffer): Unit = {
+ other.frequencyMap.asScala.foreach((keyValuePair: (Vector[UTF8String], Double)) => {
+ val key = keyValuePair._1
+ val value = keyValuePair._2
+ val originalValue = frequencyMap.getOrDefault(key, 0.0D)
+ frequencyMap.put(key, originalValue + value)
+ })
+ }
+
+ def sortWithTwoFields(frequencyDescend: Boolean)
+ (keyWithFrequency: Tuple2[Vector[UTF8String], Double],
+ keyWithFrequency2: Tuple2[Vector[UTF8String], Double]): Boolean = {
+ if (keyWithFrequency._2 != keyWithFrequency2._2) {
+ (keyWithFrequency._2 < keyWithFrequency2._2) ^ frequencyDescend
+ }
+ else {
+ val keyVector = keyWithFrequency._1
+ val keyVector2 = keyWithFrequency2._1
+ for (i <- 0 until keyVector.length) {
+ if (keyVector(i) != keyVector2(i)) {
+ return (keyVector(i).compare(keyVector2(i))) < 0
+ }
+ }
+ true
+ }
+ }
+
+ def trim(): Unit = {
+ if (frequencyMap.size() > 2 * k * precisionFactor) {
+ val orderedWithIndex = frequencyMap.asScala.iterator.toVector.
+ sortWith(sortWithTwoFields(frequencyDescend = false)).zipWithIndex
+ orderedWithIndex.takeWhile(_._2 < frequencyMap.size() - k * precisionFactor).map(_._1).
+ foreach(keyValuePair => frequencyMap.remove(keyValuePair._1))
+ }
+ }
+
+ def getTopKNGrams(): Seq[(Vector[UTF8String], Double)] = {
+ frequencyMap.asScala.iterator.toVector.sortWith(sortWithTwoFields(frequencyDescend = true)).
+ zipWithIndex.takeWhile(_._2 < k).map(_._1)
--- End diff --
You can simply do:
frequencyMap.asScala.iterator.toVector.sortWith(sortWithTwoFields(frequencyDescend = true)).take(k)
---
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 #17359: Add aggreagate expression nGrams
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/17359
Can one of the admins verify this patch?
---
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 #17359: [SPARK-20028][SQL] Add aggreagate expression nGra...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/17359#discussion_r107347252
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/NGrams.scala ---
@@ -0,0 +1,258 @@
+/*
+ * 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 java.util.HashMap
+
+import org.apache.spark.serializer.KryoSerializer
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult._
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription, ImplicitCastInputTypes, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.NGrams.NGramBuffer
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData}
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Return the top-k n-grams in rows that consist of sequences of strings.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(expr, n, k, accuracy) - Estimates the top-k n-grams in rows that consist of sequences
+ of strings, represented as arrays of strings, or arrays of arrays of strings. 'accuracy' is an
+ optional precision factor that controls memory usage.
+ The parameter 'n' specifies what type of n-grams are being estimated. Unigrams are n = 1, and
+ bigrams are n = 2. Generally, n will not be greater than about 5. The 'k' parameter specifies
+ how many of the highest-frequency n-grams will be returned by the UDAF. The optional precision
+ factor 'accuracy' specifies how much memory to use for estimation; more memory will give
+ more accurate frequency counts, but could crash the JVM. The value will be the max between
+ 'accuracy'(0 if it's not specified) and 1000/k, which indicates the max number of n-grams
+ which are kept in the internal HashMap.
+ The output is an array of maps with the top-k n-grams and corresponding frequency.
+ """,
+ extended = """
+ Examples:
+ > SELECT ngrams(array("abc", "abc", "bcd", "abc", "bcd"), 2, 4);
+ [{["abc","bcd"]:2.0},
+ {["abc","abc"]:1.0},
+ {["bcd","abc"]:1.0}]
+ """)
+case class NGrams(
+ child: Expression,
+ nExpression: Expression,
+ kExpression: Expression,
+ accuracyExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int)
+ extends TypedImperativeAggregate[NGramBuffer] with ImplicitCastInputTypes {
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression,
+ accuracyExpression: Expression) = {
+ this(child, nExpression, kExpression, NGrams.getAccuracy(kExpression, accuracyExpression), 0, 0)
+ }
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression) = {
+ this(child, nExpression, kExpression, Literal(0))
+ }
+
+ private lazy val n: Int = nExpression.eval().asInstanceOf[Int]
+ private lazy val k: Int = kExpression.eval().asInstanceOf[Int]
+ private lazy val accuracy: Int = accuracyExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(ArrayType(StringType, false), ArrayType(ArrayType(StringType, false))),
+ IntegerType, IntegerType, IntegerType)
+ }
+
+ val isArrayOfString = child.dataType == ArrayType(StringType, false) ||
+ child.dataType == ArrayType(StringType, true)
--- End diff --
Are you meaning `ArrayType(ArrayType(StringType, false))` here?
---
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 #17359: [SPARK-20028][SQL] Add aggreagate expression nGrams
Posted by chenghao-intel <gi...@git.apache.org>.
Github user chenghao-intel commented on the issue:
https://github.com/apache/spark/pull/17359
@rxin nGram is the built-in UDAF in Hive, and some users complaints they faced performance issue when running the queries with nGram.
---
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 #17359: [SPARK-20028][SQL] Add aggreagate expression nGra...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/17359#discussion_r107347661
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/NGrams.scala ---
@@ -0,0 +1,258 @@
+/*
+ * 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 java.util.HashMap
+
+import org.apache.spark.serializer.KryoSerializer
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult._
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription, ImplicitCastInputTypes, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.NGrams.NGramBuffer
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData}
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Return the top-k n-grams in rows that consist of sequences of strings.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(expr, n, k, accuracy) - Estimates the top-k n-grams in rows that consist of sequences
+ of strings, represented as arrays of strings, or arrays of arrays of strings. 'accuracy' is an
+ optional precision factor that controls memory usage.
+ The parameter 'n' specifies what type of n-grams are being estimated. Unigrams are n = 1, and
+ bigrams are n = 2. Generally, n will not be greater than about 5. The 'k' parameter specifies
+ how many of the highest-frequency n-grams will be returned by the UDAF. The optional precision
+ factor 'accuracy' specifies how much memory to use for estimation; more memory will give
+ more accurate frequency counts, but could crash the JVM. The value will be the max between
+ 'accuracy'(0 if it's not specified) and 1000/k, which indicates the max number of n-grams
+ which are kept in the internal HashMap.
+ The output is an array of maps with the top-k n-grams and corresponding frequency.
+ """,
+ extended = """
+ Examples:
+ > SELECT ngrams(array("abc", "abc", "bcd", "abc", "bcd"), 2, 4);
+ [{["abc","bcd"]:2.0},
+ {["abc","abc"]:1.0},
+ {["bcd","abc"]:1.0}]
+ """)
+case class NGrams(
+ child: Expression,
+ nExpression: Expression,
+ kExpression: Expression,
+ accuracyExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int)
+ extends TypedImperativeAggregate[NGramBuffer] with ImplicitCastInputTypes {
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression,
+ accuracyExpression: Expression) = {
+ this(child, nExpression, kExpression, NGrams.getAccuracy(kExpression, accuracyExpression), 0, 0)
+ }
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression) = {
+ this(child, nExpression, kExpression, Literal(0))
+ }
+
+ private lazy val n: Int = nExpression.eval().asInstanceOf[Int]
+ private lazy val k: Int = kExpression.eval().asInstanceOf[Int]
+ private lazy val accuracy: Int = accuracyExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(ArrayType(StringType, false), ArrayType(ArrayType(StringType, false))),
+ IntegerType, IntegerType, IntegerType)
+ }
+
+ val isArrayOfString = child.dataType == ArrayType(StringType, false) ||
+ child.dataType == ArrayType(StringType, true)
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (!nExpression.foldable || !kExpression.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)")
--- End diff --
This is not type check and failure. Shall we check `accuracy` when initialized `NGrams` and throw analysis error?
---
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 #17359: [SPARK-20028][SQL] Add aggreagate expression nGra...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/17359#discussion_r107353837
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/NGrams.scala ---
@@ -0,0 +1,258 @@
+/*
+ * 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 java.util.HashMap
+
+import org.apache.spark.serializer.KryoSerializer
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult._
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription, ImplicitCastInputTypes, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.NGrams.NGramBuffer
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData}
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Return the top-k n-grams in rows that consist of sequences of strings.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(expr, n, k, accuracy) - Estimates the top-k n-grams in rows that consist of sequences
+ of strings, represented as arrays of strings, or arrays of arrays of strings. 'accuracy' is an
+ optional precision factor that controls memory usage.
+ The parameter 'n' specifies what type of n-grams are being estimated. Unigrams are n = 1, and
+ bigrams are n = 2. Generally, n will not be greater than about 5. The 'k' parameter specifies
+ how many of the highest-frequency n-grams will be returned by the UDAF. The optional precision
+ factor 'accuracy' specifies how much memory to use for estimation; more memory will give
+ more accurate frequency counts, but could crash the JVM. The value will be the max between
+ 'accuracy'(0 if it's not specified) and 1000/k, which indicates the max number of n-grams
+ which are kept in the internal HashMap.
+ The output is an array of maps with the top-k n-grams and corresponding frequency.
+ """,
+ extended = """
+ Examples:
+ > SELECT ngrams(array("abc", "abc", "bcd", "abc", "bcd"), 2, 4);
+ [{["abc","bcd"]:2.0},
+ {["abc","abc"]:1.0},
+ {["bcd","abc"]:1.0}]
+ """)
+case class NGrams(
+ child: Expression,
+ nExpression: Expression,
+ kExpression: Expression,
+ accuracyExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int)
+ extends TypedImperativeAggregate[NGramBuffer] with ImplicitCastInputTypes {
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression,
+ accuracyExpression: Expression) = {
+ this(child, nExpression, kExpression, NGrams.getAccuracy(kExpression, accuracyExpression), 0, 0)
+ }
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression) = {
+ this(child, nExpression, kExpression, Literal(0))
+ }
+
+ private lazy val n: Int = nExpression.eval().asInstanceOf[Int]
+ private lazy val k: Int = kExpression.eval().asInstanceOf[Int]
+ private lazy val accuracy: Int = accuracyExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(ArrayType(StringType, false), ArrayType(ArrayType(StringType, false))),
+ IntegerType, IntegerType, IntegerType)
+ }
+
+ val isArrayOfString = child.dataType == ArrayType(StringType, false) ||
+ child.dataType == ArrayType(StringType, true)
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (!nExpression.foldable || !kExpression.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 {
+ TypeCheckSuccess
+ }
+ }
+
+ override def createAggregationBuffer(): NGramBuffer = {
+ new NGramBuffer(n, k, accuracy, new HashMap[Vector[UTF8String], Double]())
+ }
+
+ def updateArray(genericArrayData: GenericArrayData, buffer: NGramBuffer, inputRow: InternalRow) {
+ val values = (0 until genericArrayData.numElements()).map(genericArrayData.get(_, StringType).
+ asInstanceOf[UTF8String]).toVector
+ val nGrams = getNGrams(values, n)
+ nGrams.foreach(buffer.add(_))
+ buffer.trim()
+ }
+ override def update(buffer: NGramBuffer, inputRow: InternalRow): NGramBuffer = {
+ if (isArrayOfString) {
+ updateArray(child.eval(inputRow).asInstanceOf[GenericArrayData], buffer, inputRow)
+ }
+ else {
+ val arrayOfArray = child.eval(inputRow).asInstanceOf[GenericArrayData]
+ for (i <- 0 until arrayOfArray.numElements()) {
+ updateArray(arrayOfArray.getArray(i).asInstanceOf[GenericArrayData], buffer, inputRow)
+ }
+ }
+ buffer
+ }
+
+ override def merge(buffer: NGramBuffer, input: NGramBuffer): NGramBuffer = {
+ buffer.merge(input)
+ buffer.trim()
+ buffer
+ }
+
+ override def eval(buffer: NGramBuffer): Any = {
+ val topKNGrams = buffer.getTopKNGrams().map((keyValuePair: (Vector[UTF8String], Double)) => {
+ val arrayKey = new GenericArrayData(Vector(new GenericArrayData(keyValuePair._1)))
+ val arrayValue = new GenericArrayData(Vector(keyValuePair._2))
+ new ArrayBasedMapData(arrayKey, arrayValue)
+ }).toVector
+ new GenericArrayData(topKNGrams)
--- End diff --
wrong ident. There are many style issues. Can you run style check locally and fix them?
---
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 #17359: [SPARK-20028][SQL] Add aggreagate expression nGra...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/17359#discussion_r107350936
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/NGrams.scala ---
@@ -0,0 +1,258 @@
+/*
+ * 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 java.util.HashMap
+
+import org.apache.spark.serializer.KryoSerializer
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult._
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription, ImplicitCastInputTypes, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.NGrams.NGramBuffer
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData}
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Return the top-k n-grams in rows that consist of sequences of strings.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(expr, n, k, accuracy) - Estimates the top-k n-grams in rows that consist of sequences
+ of strings, represented as arrays of strings, or arrays of arrays of strings. 'accuracy' is an
+ optional precision factor that controls memory usage.
+ The parameter 'n' specifies what type of n-grams are being estimated. Unigrams are n = 1, and
+ bigrams are n = 2. Generally, n will not be greater than about 5. The 'k' parameter specifies
+ how many of the highest-frequency n-grams will be returned by the UDAF. The optional precision
+ factor 'accuracy' specifies how much memory to use for estimation; more memory will give
+ more accurate frequency counts, but could crash the JVM. The value will be the max between
+ 'accuracy'(0 if it's not specified) and 1000/k, which indicates the max number of n-grams
+ which are kept in the internal HashMap.
+ The output is an array of maps with the top-k n-grams and corresponding frequency.
+ """,
+ extended = """
+ Examples:
+ > SELECT ngrams(array("abc", "abc", "bcd", "abc", "bcd"), 2, 4);
+ [{["abc","bcd"]:2.0},
+ {["abc","abc"]:1.0},
+ {["bcd","abc"]:1.0}]
+ """)
+case class NGrams(
+ child: Expression,
+ nExpression: Expression,
+ kExpression: Expression,
+ accuracyExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int)
+ extends TypedImperativeAggregate[NGramBuffer] with ImplicitCastInputTypes {
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression,
+ accuracyExpression: Expression) = {
+ this(child, nExpression, kExpression, NGrams.getAccuracy(kExpression, accuracyExpression), 0, 0)
+ }
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression) = {
+ this(child, nExpression, kExpression, Literal(0))
+ }
+
+ private lazy val n: Int = nExpression.eval().asInstanceOf[Int]
+ private lazy val k: Int = kExpression.eval().asInstanceOf[Int]
+ private lazy val accuracy: Int = accuracyExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(ArrayType(StringType, false), ArrayType(ArrayType(StringType, false))),
+ IntegerType, IntegerType, IntegerType)
+ }
+
+ val isArrayOfString = child.dataType == ArrayType(StringType, false) ||
+ child.dataType == ArrayType(StringType, true)
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (!nExpression.foldable || !kExpression.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 {
+ TypeCheckSuccess
+ }
+ }
+
+ override def createAggregationBuffer(): NGramBuffer = {
+ new NGramBuffer(n, k, accuracy, new HashMap[Vector[UTF8String], Double]())
+ }
+
+ def updateArray(genericArrayData: GenericArrayData, buffer: NGramBuffer, inputRow: InternalRow) {
+ val values = (0 until genericArrayData.numElements()).map(genericArrayData.get(_, StringType).
+ asInstanceOf[UTF8String]).toVector
+ val nGrams = getNGrams(values, n)
+ nGrams.foreach(buffer.add(_))
+ buffer.trim()
+ }
+ override def update(buffer: NGramBuffer, inputRow: InternalRow): NGramBuffer = {
+ if (isArrayOfString) {
+ updateArray(child.eval(inputRow).asInstanceOf[GenericArrayData], buffer, inputRow)
+ }
+ else {
+ val arrayOfArray = child.eval(inputRow).asInstanceOf[GenericArrayData]
+ for (i <- 0 until arrayOfArray.numElements()) {
+ updateArray(arrayOfArray.getArray(i).asInstanceOf[GenericArrayData], buffer, inputRow)
+ }
+ }
+ buffer
+ }
+
+ override def merge(buffer: NGramBuffer, input: NGramBuffer): NGramBuffer = {
+ buffer.merge(input)
+ buffer.trim()
+ buffer
+ }
+
+ override def eval(buffer: NGramBuffer): Any = {
+ val topKNGrams = buffer.getTopKNGrams().map((keyValuePair: (Vector[UTF8String], Double)) => {
+ val arrayKey = new GenericArrayData(Vector(new GenericArrayData(keyValuePair._1)))
+ val arrayValue = new GenericArrayData(Vector(keyValuePair._2))
+ new ArrayBasedMapData(arrayKey, arrayValue)
+ }).toVector
+ new GenericArrayData(topKNGrams)
+ }
+
+ private def getNGrams(values: Vector[UTF8String], n: Int): Vector[Vector[UTF8String]] = {
+ if (values.length >= n) {
+ values.sliding(n).toVector
+ }
+ else {
+ Vector()
+ }
+ }
+
+ override def withNewMutableAggBufferOffset(newOffset: Int): NGrams =
+ copy(mutableAggBufferOffset = newOffset)
+
+ override def withNewInputAggBufferOffset(newOffset: Int): NGrams =
+ copy(inputAggBufferOffset = newOffset)
+
+ override def children: Seq[Expression] = Seq(child, nExpression, kExpression, accuracyExpression)
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = ArrayType(MapType(ArrayType(StringType), DoubleType))
+
+ override def prettyName: String = "ngrams"
+
+ override def serialize(obj: NGramBuffer): Array[Byte] = {
+ NGrams.serializer.serialize(obj)
+ }
+
+ override def deserialize(bytes: Array[Byte]): NGramBuffer = {
+ NGrams.serializer.deserialize(bytes)
+ }
+}
+
+object NGrams {
+
+ val DEFAULT_ACCURACY: Int = 1000
+
+ private def getAccuracy(kExpression: Expression, accuracyExpression: Expression): Expression = {
+ val accuracy = accuracyExpression.eval().asInstanceOf[Int]
+ val k = kExpression.eval().asInstanceOf[Int]
+ Literal(accuracy.max(DEFAULT_ACCURACY / k))
+ }
+
+ val kryoSerializer: KryoSerializer = new KryoSerializer(new SparkConf())
+
+ import collection.JavaConverters._
+
+ class NGramBuffer(val n: Int,
+ val k: Int,
+ val precisionFactor: Int,
+ val frequencyMap: HashMap[Vector[UTF8String], Double]) {
+ def add(ng: Vector[UTF8String]): Unit = {
+ var currentFrequency: Double = frequencyMap.get(ng)
+ if (currentFrequency == null.asInstanceOf[Double]) {
+ currentFrequency = 1.0D
+ } else {
+ currentFrequency += 1
+ }
+ frequencyMap.put(ng, currentFrequency)
+ }
+
+ def merge(other: NGramBuffer): Unit = {
+ other.frequencyMap.asScala.foreach((keyValuePair: (Vector[UTF8String], Double)) => {
+ val key = keyValuePair._1
+ val value = keyValuePair._2
+ val originalValue = frequencyMap.getOrDefault(key, 0.0D)
+ frequencyMap.put(key, originalValue + value)
--- End diff --
Oh. you use java's HashMap.
But `getOrDefault` is only supported in Java8. I think we haven't drop Java7 support for now.
---
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 #17359: [SPARK-20028][SQL] Add aggreagate expression nGra...
Posted by gczsjdy <gi...@git.apache.org>.
Github user gczsjdy commented on a diff in the pull request:
https://github.com/apache/spark/pull/17359#discussion_r107347663
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/NGrams.scala ---
@@ -0,0 +1,258 @@
+/*
+ * 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 java.util.HashMap
+
+import org.apache.spark.serializer.KryoSerializer
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult._
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription, ImplicitCastInputTypes, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.NGrams.NGramBuffer
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData}
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Return the top-k n-grams in rows that consist of sequences of strings.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(expr, n, k, accuracy) - Estimates the top-k n-grams in rows that consist of sequences
+ of strings, represented as arrays of strings, or arrays of arrays of strings. 'accuracy' is an
--- End diff --
Yes, please refer to https://cwiki.apache.org/confluence/display/Hive/LanguageManual+UDF and https://cwiki.apache.org/confluence/display/Hive/StatisticsAndDataMining#StatisticsAndDataMining-Usage
---
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 #17359: [SPARK-20028][SQL] Add aggreagate expression nGra...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/17359#discussion_r107349720
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/NGrams.scala ---
@@ -0,0 +1,258 @@
+/*
+ * 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 java.util.HashMap
+
+import org.apache.spark.serializer.KryoSerializer
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult._
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription, ImplicitCastInputTypes, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.NGrams.NGramBuffer
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData}
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Return the top-k n-grams in rows that consist of sequences of strings.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(expr, n, k, accuracy) - Estimates the top-k n-grams in rows that consist of sequences
+ of strings, represented as arrays of strings, or arrays of arrays of strings. 'accuracy' is an
+ optional precision factor that controls memory usage.
+ The parameter 'n' specifies what type of n-grams are being estimated. Unigrams are n = 1, and
+ bigrams are n = 2. Generally, n will not be greater than about 5. The 'k' parameter specifies
+ how many of the highest-frequency n-grams will be returned by the UDAF. The optional precision
+ factor 'accuracy' specifies how much memory to use for estimation; more memory will give
+ more accurate frequency counts, but could crash the JVM. The value will be the max between
+ 'accuracy'(0 if it's not specified) and 1000/k, which indicates the max number of n-grams
+ which are kept in the internal HashMap.
+ The output is an array of maps with the top-k n-grams and corresponding frequency.
+ """,
+ extended = """
+ Examples:
+ > SELECT ngrams(array("abc", "abc", "bcd", "abc", "bcd"), 2, 4);
+ [{["abc","bcd"]:2.0},
+ {["abc","abc"]:1.0},
+ {["bcd","abc"]:1.0}]
+ """)
+case class NGrams(
+ child: Expression,
+ nExpression: Expression,
+ kExpression: Expression,
+ accuracyExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int)
+ extends TypedImperativeAggregate[NGramBuffer] with ImplicitCastInputTypes {
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression,
+ accuracyExpression: Expression) = {
+ this(child, nExpression, kExpression, NGrams.getAccuracy(kExpression, accuracyExpression), 0, 0)
+ }
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression) = {
+ this(child, nExpression, kExpression, Literal(0))
+ }
+
+ private lazy val n: Int = nExpression.eval().asInstanceOf[Int]
--- End diff --
n can't be <= 0.
---
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 #17359: [SPARK-20028][SQL] Add aggreagate expression nGra...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/17359#discussion_r107349865
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/NGrams.scala ---
@@ -0,0 +1,258 @@
+/*
+ * 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 java.util.HashMap
+
+import org.apache.spark.serializer.KryoSerializer
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult._
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription, ImplicitCastInputTypes, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.NGrams.NGramBuffer
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData}
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Return the top-k n-grams in rows that consist of sequences of strings.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(expr, n, k, accuracy) - Estimates the top-k n-grams in rows that consist of sequences
+ of strings, represented as arrays of strings, or arrays of arrays of strings. 'accuracy' is an
+ optional precision factor that controls memory usage.
+ The parameter 'n' specifies what type of n-grams are being estimated. Unigrams are n = 1, and
+ bigrams are n = 2. Generally, n will not be greater than about 5. The 'k' parameter specifies
+ how many of the highest-frequency n-grams will be returned by the UDAF. The optional precision
+ factor 'accuracy' specifies how much memory to use for estimation; more memory will give
+ more accurate frequency counts, but could crash the JVM. The value will be the max between
+ 'accuracy'(0 if it's not specified) and 1000/k, which indicates the max number of n-grams
+ which are kept in the internal HashMap.
+ The output is an array of maps with the top-k n-grams and corresponding frequency.
+ """,
+ extended = """
+ Examples:
+ > SELECT ngrams(array("abc", "abc", "bcd", "abc", "bcd"), 2, 4);
+ [{["abc","bcd"]:2.0},
+ {["abc","abc"]:1.0},
+ {["bcd","abc"]:1.0}]
+ """)
+case class NGrams(
+ child: Expression,
+ nExpression: Expression,
+ kExpression: Expression,
+ accuracyExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int)
+ extends TypedImperativeAggregate[NGramBuffer] with ImplicitCastInputTypes {
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression,
+ accuracyExpression: Expression) = {
+ this(child, nExpression, kExpression, NGrams.getAccuracy(kExpression, accuracyExpression), 0, 0)
+ }
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression) = {
+ this(child, nExpression, kExpression, Literal(0))
+ }
+
+ private lazy val n: Int = nExpression.eval().asInstanceOf[Int]
+ private lazy val k: Int = kExpression.eval().asInstanceOf[Int]
+ private lazy val accuracy: Int = accuracyExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(ArrayType(StringType, false), ArrayType(ArrayType(StringType, false))),
+ IntegerType, IntegerType, IntegerType)
+ }
+
+ val isArrayOfString = child.dataType == ArrayType(StringType, false) ||
+ child.dataType == ArrayType(StringType, true)
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (!nExpression.foldable || !kExpression.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 {
+ TypeCheckSuccess
+ }
+ }
+
+ override def createAggregationBuffer(): NGramBuffer = {
+ new NGramBuffer(n, k, accuracy, new HashMap[Vector[UTF8String], Double]())
+ }
+
+ def updateArray(genericArrayData: GenericArrayData, buffer: NGramBuffer, inputRow: InternalRow) {
+ val values = (0 until genericArrayData.numElements()).map(genericArrayData.get(_, StringType).
+ asInstanceOf[UTF8String]).toVector
+ val nGrams = getNGrams(values, n)
+ nGrams.foreach(buffer.add(_))
+ buffer.trim()
+ }
+ override def update(buffer: NGramBuffer, inputRow: InternalRow): NGramBuffer = {
+ if (isArrayOfString) {
+ updateArray(child.eval(inputRow).asInstanceOf[GenericArrayData], buffer, inputRow)
+ }
+ else {
+ val arrayOfArray = child.eval(inputRow).asInstanceOf[GenericArrayData]
+ for (i <- 0 until arrayOfArray.numElements()) {
+ updateArray(arrayOfArray.getArray(i).asInstanceOf[GenericArrayData], buffer, inputRow)
+ }
+ }
+ buffer
+ }
+
+ override def merge(buffer: NGramBuffer, input: NGramBuffer): NGramBuffer = {
+ buffer.merge(input)
+ buffer.trim()
+ buffer
+ }
+
+ override def eval(buffer: NGramBuffer): Any = {
+ val topKNGrams = buffer.getTopKNGrams().map((keyValuePair: (Vector[UTF8String], Double)) => {
+ val arrayKey = new GenericArrayData(Vector(new GenericArrayData(keyValuePair._1)))
+ val arrayValue = new GenericArrayData(Vector(keyValuePair._2))
+ new ArrayBasedMapData(arrayKey, arrayValue)
+ }).toVector
+ new GenericArrayData(topKNGrams)
+ }
+
+ private def getNGrams(values: Vector[UTF8String], n: Int): Vector[Vector[UTF8String]] = {
+ if (values.length >= n) {
+ values.sliding(n).toVector
+ }
--- End diff --
style:
if {
...
} else {
...
}
---
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 #17359: [SPARK-20028][SQL] Add aggreagate expression nGra...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/17359#discussion_r107348550
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/NGrams.scala ---
@@ -0,0 +1,258 @@
+/*
+ * 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 java.util.HashMap
+
+import org.apache.spark.serializer.KryoSerializer
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult._
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription, ImplicitCastInputTypes, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.NGrams.NGramBuffer
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData}
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Return the top-k n-grams in rows that consist of sequences of strings.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(expr, n, k, accuracy) - Estimates the top-k n-grams in rows that consist of sequences
+ of strings, represented as arrays of strings, or arrays of arrays of strings. 'accuracy' is an
+ optional precision factor that controls memory usage.
+ The parameter 'n' specifies what type of n-grams are being estimated. Unigrams are n = 1, and
+ bigrams are n = 2. Generally, n will not be greater than about 5. The 'k' parameter specifies
+ how many of the highest-frequency n-grams will be returned by the UDAF. The optional precision
+ factor 'accuracy' specifies how much memory to use for estimation; more memory will give
+ more accurate frequency counts, but could crash the JVM. The value will be the max between
+ 'accuracy'(0 if it's not specified) and 1000/k, which indicates the max number of n-grams
+ which are kept in the internal HashMap.
+ The output is an array of maps with the top-k n-grams and corresponding frequency.
+ """,
+ extended = """
+ Examples:
+ > SELECT ngrams(array("abc", "abc", "bcd", "abc", "bcd"), 2, 4);
+ [{["abc","bcd"]:2.0},
+ {["abc","abc"]:1.0},
+ {["bcd","abc"]:1.0}]
+ """)
+case class NGrams(
+ child: Expression,
+ nExpression: Expression,
+ kExpression: Expression,
+ accuracyExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int)
+ extends TypedImperativeAggregate[NGramBuffer] with ImplicitCastInputTypes {
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression,
+ accuracyExpression: Expression) = {
+ this(child, nExpression, kExpression, NGrams.getAccuracy(kExpression, accuracyExpression), 0, 0)
+ }
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression) = {
+ this(child, nExpression, kExpression, Literal(0))
+ }
+
+ private lazy val n: Int = nExpression.eval().asInstanceOf[Int]
+ private lazy val k: Int = kExpression.eval().asInstanceOf[Int]
+ private lazy val accuracy: Int = accuracyExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(ArrayType(StringType, false), ArrayType(ArrayType(StringType, false))),
+ IntegerType, IntegerType, IntegerType)
+ }
+
+ val isArrayOfString = child.dataType == ArrayType(StringType, false) ||
+ child.dataType == ArrayType(StringType, true)
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (!nExpression.foldable || !kExpression.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 {
+ TypeCheckSuccess
+ }
+ }
+
+ override def createAggregationBuffer(): NGramBuffer = {
+ new NGramBuffer(n, k, accuracy, new HashMap[Vector[UTF8String], Double]())
+ }
+
+ def updateArray(genericArrayData: GenericArrayData, buffer: NGramBuffer, inputRow: InternalRow) {
+ val values = (0 until genericArrayData.numElements()).map(genericArrayData.get(_, StringType).
+ asInstanceOf[UTF8String]).toVector
+ val nGrams = getNGrams(values, n)
+ nGrams.foreach(buffer.add(_))
+ buffer.trim()
+ }
+ override def update(buffer: NGramBuffer, inputRow: InternalRow): NGramBuffer = {
+ if (isArrayOfString) {
+ updateArray(child.eval(inputRow).asInstanceOf[GenericArrayData], buffer, inputRow)
+ }
+ else {
+ val arrayOfArray = child.eval(inputRow).asInstanceOf[GenericArrayData]
+ for (i <- 0 until arrayOfArray.numElements()) {
+ updateArray(arrayOfArray.getArray(i).asInstanceOf[GenericArrayData], buffer, inputRow)
+ }
+ }
+ buffer
+ }
+
+ override def merge(buffer: NGramBuffer, input: NGramBuffer): NGramBuffer = {
+ buffer.merge(input)
+ buffer.trim()
+ buffer
+ }
+
+ override def eval(buffer: NGramBuffer): Any = {
+ val topKNGrams = buffer.getTopKNGrams().map((keyValuePair: (Vector[UTF8String], Double)) => {
+ val arrayKey = new GenericArrayData(Vector(new GenericArrayData(keyValuePair._1)))
+ val arrayValue = new GenericArrayData(Vector(keyValuePair._2))
+ new ArrayBasedMapData(arrayKey, arrayValue)
+ }).toVector
+ new GenericArrayData(topKNGrams)
+ }
+
+ private def getNGrams(values: Vector[UTF8String], n: Int): Vector[Vector[UTF8String]] = {
+ if (values.length >= n) {
+ values.sliding(n).toVector
+ }
+ else {
+ Vector()
+ }
+ }
+
+ override def withNewMutableAggBufferOffset(newOffset: Int): NGrams =
+ copy(mutableAggBufferOffset = newOffset)
+
+ override def withNewInputAggBufferOffset(newOffset: Int): NGrams =
+ copy(inputAggBufferOffset = newOffset)
+
+ override def children: Seq[Expression] = Seq(child, nExpression, kExpression, accuracyExpression)
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = ArrayType(MapType(ArrayType(StringType), DoubleType))
+
+ override def prettyName: String = "ngrams"
+
+ override def serialize(obj: NGramBuffer): Array[Byte] = {
+ NGrams.serializer.serialize(obj)
+ }
+
+ override def deserialize(bytes: Array[Byte]): NGramBuffer = {
+ NGrams.serializer.deserialize(bytes)
+ }
+}
+
+object NGrams {
+
+ val DEFAULT_ACCURACY: Int = 1000
+
+ private def getAccuracy(kExpression: Expression, accuracyExpression: Expression): Expression = {
+ val accuracy = accuracyExpression.eval().asInstanceOf[Int]
+ val k = kExpression.eval().asInstanceOf[Int]
+ Literal(accuracy.max(DEFAULT_ACCURACY / k))
+ }
+
+ val kryoSerializer: KryoSerializer = new KryoSerializer(new SparkConf())
+
+ import collection.JavaConverters._
+
+ class NGramBuffer(val n: Int,
+ val k: Int,
+ val precisionFactor: Int,
+ val frequencyMap: HashMap[Vector[UTF8String], Double]) {
+ def add(ng: Vector[UTF8String]): Unit = {
+ var currentFrequency: Double = frequencyMap.get(ng)
+ if (currentFrequency == null.asInstanceOf[Double]) {
+ currentFrequency = 1.0D
+ } else {
+ currentFrequency += 1
+ }
+ frequencyMap.put(ng, currentFrequency)
+ }
+
+ def merge(other: NGramBuffer): Unit = {
+ other.frequencyMap.asScala.foreach((keyValuePair: (Vector[UTF8String], Double)) => {
+ val key = keyValuePair._1
+ val value = keyValuePair._2
+ val originalValue = frequencyMap.getOrDefault(key, 0.0D)
+ frequencyMap.put(key, originalValue + value)
--- End diff --
frequencyMap.put(key, frequencyMap.getOrElse(key, 0.0D) + value)
---
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 #17359: [SPARK-20028][SQL] Add aggreagate expression nGra...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/17359#discussion_r107349156
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/NGrams.scala ---
@@ -0,0 +1,258 @@
+/*
+ * 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 java.util.HashMap
+
+import org.apache.spark.serializer.KryoSerializer
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult._
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription, ImplicitCastInputTypes, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.NGrams.NGramBuffer
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData}
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Return the top-k n-grams in rows that consist of sequences of strings.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(expr, n, k, accuracy) - Estimates the top-k n-grams in rows that consist of sequences
+ of strings, represented as arrays of strings, or arrays of arrays of strings. 'accuracy' is an
+ optional precision factor that controls memory usage.
+ The parameter 'n' specifies what type of n-grams are being estimated. Unigrams are n = 1, and
+ bigrams are n = 2. Generally, n will not be greater than about 5. The 'k' parameter specifies
+ how many of the highest-frequency n-grams will be returned by the UDAF. The optional precision
+ factor 'accuracy' specifies how much memory to use for estimation; more memory will give
+ more accurate frequency counts, but could crash the JVM. The value will be the max between
+ 'accuracy'(0 if it's not specified) and 1000/k, which indicates the max number of n-grams
+ which are kept in the internal HashMap.
+ The output is an array of maps with the top-k n-grams and corresponding frequency.
+ """,
+ extended = """
+ Examples:
+ > SELECT ngrams(array("abc", "abc", "bcd", "abc", "bcd"), 2, 4);
+ [{["abc","bcd"]:2.0},
+ {["abc","abc"]:1.0},
+ {["bcd","abc"]:1.0}]
+ """)
+case class NGrams(
+ child: Expression,
+ nExpression: Expression,
+ kExpression: Expression,
+ accuracyExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int)
+ extends TypedImperativeAggregate[NGramBuffer] with ImplicitCastInputTypes {
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression,
+ accuracyExpression: Expression) = {
+ this(child, nExpression, kExpression, NGrams.getAccuracy(kExpression, accuracyExpression), 0, 0)
+ }
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression) = {
+ this(child, nExpression, kExpression, Literal(0))
+ }
+
+ private lazy val n: Int = nExpression.eval().asInstanceOf[Int]
+ private lazy val k: Int = kExpression.eval().asInstanceOf[Int]
+ private lazy val accuracy: Int = accuracyExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(ArrayType(StringType, false), ArrayType(ArrayType(StringType, false))),
+ IntegerType, IntegerType, IntegerType)
+ }
+
+ val isArrayOfString = child.dataType == ArrayType(StringType, false) ||
+ child.dataType == ArrayType(StringType, true)
--- End diff --
`ArrayType(StringType, true)` is not included in `inputTypes`?
---
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 #17359: [SPARK-20028][SQL] Add aggreagate expression nGra...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/17359#discussion_r107350129
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/NGrams.scala ---
@@ -0,0 +1,258 @@
+/*
+ * 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 java.util.HashMap
+
+import org.apache.spark.serializer.KryoSerializer
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult._
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription, ImplicitCastInputTypes, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.NGrams.NGramBuffer
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData}
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Return the top-k n-grams in rows that consist of sequences of strings.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(expr, n, k, accuracy) - Estimates the top-k n-grams in rows that consist of sequences
+ of strings, represented as arrays of strings, or arrays of arrays of strings. 'accuracy' is an
+ optional precision factor that controls memory usage.
+ The parameter 'n' specifies what type of n-grams are being estimated. Unigrams are n = 1, and
+ bigrams are n = 2. Generally, n will not be greater than about 5. The 'k' parameter specifies
+ how many of the highest-frequency n-grams will be returned by the UDAF. The optional precision
+ factor 'accuracy' specifies how much memory to use for estimation; more memory will give
+ more accurate frequency counts, but could crash the JVM. The value will be the max between
+ 'accuracy'(0 if it's not specified) and 1000/k, which indicates the max number of n-grams
+ which are kept in the internal HashMap.
+ The output is an array of maps with the top-k n-grams and corresponding frequency.
+ """,
+ extended = """
+ Examples:
+ > SELECT ngrams(array("abc", "abc", "bcd", "abc", "bcd"), 2, 4);
+ [{["abc","bcd"]:2.0},
+ {["abc","abc"]:1.0},
+ {["bcd","abc"]:1.0}]
+ """)
+case class NGrams(
+ child: Expression,
+ nExpression: Expression,
+ kExpression: Expression,
+ accuracyExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int)
+ extends TypedImperativeAggregate[NGramBuffer] with ImplicitCastInputTypes {
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression,
+ accuracyExpression: Expression) = {
+ this(child, nExpression, kExpression, NGrams.getAccuracy(kExpression, accuracyExpression), 0, 0)
+ }
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression) = {
+ this(child, nExpression, kExpression, Literal(0))
+ }
+
+ private lazy val n: Int = nExpression.eval().asInstanceOf[Int]
+ private lazy val k: Int = kExpression.eval().asInstanceOf[Int]
+ private lazy val accuracy: Int = accuracyExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(ArrayType(StringType, false), ArrayType(ArrayType(StringType, false))),
+ IntegerType, IntegerType, IntegerType)
+ }
+
+ val isArrayOfString = child.dataType == ArrayType(StringType, false) ||
+ child.dataType == ArrayType(StringType, true)
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (!nExpression.foldable || !kExpression.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 {
+ TypeCheckSuccess
+ }
+ }
+
+ override def createAggregationBuffer(): NGramBuffer = {
+ new NGramBuffer(n, k, accuracy, new HashMap[Vector[UTF8String], Double]())
+ }
+
+ def updateArray(genericArrayData: GenericArrayData, buffer: NGramBuffer, inputRow: InternalRow) {
+ val values = (0 until genericArrayData.numElements()).map(genericArrayData.get(_, StringType).
+ asInstanceOf[UTF8String]).toVector
+ val nGrams = getNGrams(values, n)
+ nGrams.foreach(buffer.add(_))
+ buffer.trim()
+ }
+ override def update(buffer: NGramBuffer, inputRow: InternalRow): NGramBuffer = {
+ if (isArrayOfString) {
+ updateArray(child.eval(inputRow).asInstanceOf[GenericArrayData], buffer, inputRow)
+ }
+ else {
+ val arrayOfArray = child.eval(inputRow).asInstanceOf[GenericArrayData]
+ for (i <- 0 until arrayOfArray.numElements()) {
+ updateArray(arrayOfArray.getArray(i).asInstanceOf[GenericArrayData], buffer, inputRow)
+ }
+ }
+ buffer
+ }
+
+ override def merge(buffer: NGramBuffer, input: NGramBuffer): NGramBuffer = {
+ buffer.merge(input)
+ buffer.trim()
+ buffer
+ }
+
+ override def eval(buffer: NGramBuffer): Any = {
+ val topKNGrams = buffer.getTopKNGrams().map((keyValuePair: (Vector[UTF8String], Double)) => {
+ val arrayKey = new GenericArrayData(Vector(new GenericArrayData(keyValuePair._1)))
+ val arrayValue = new GenericArrayData(Vector(keyValuePair._2))
+ new ArrayBasedMapData(arrayKey, arrayValue)
+ }).toVector
+ new GenericArrayData(topKNGrams)
+ }
+
+ private def getNGrams(values: Vector[UTF8String], n: Int): Vector[Vector[UTF8String]] = {
+ if (values.length >= n) {
+ values.sliding(n).toVector
+ }
+ else {
+ Vector()
+ }
+ }
+
+ override def withNewMutableAggBufferOffset(newOffset: Int): NGrams =
+ copy(mutableAggBufferOffset = newOffset)
+
+ override def withNewInputAggBufferOffset(newOffset: Int): NGrams =
+ copy(inputAggBufferOffset = newOffset)
+
+ override def children: Seq[Expression] = Seq(child, nExpression, kExpression, accuracyExpression)
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = ArrayType(MapType(ArrayType(StringType), DoubleType))
+
+ override def prettyName: String = "ngrams"
+
+ override def serialize(obj: NGramBuffer): Array[Byte] = {
+ NGrams.serializer.serialize(obj)
+ }
+
+ override def deserialize(bytes: Array[Byte]): NGramBuffer = {
+ NGrams.serializer.deserialize(bytes)
+ }
+}
+
+object NGrams {
+
+ val DEFAULT_ACCURACY: Int = 1000
+
+ private def getAccuracy(kExpression: Expression, accuracyExpression: Expression): Expression = {
+ val accuracy = accuracyExpression.eval().asInstanceOf[Int]
+ val k = kExpression.eval().asInstanceOf[Int]
+ Literal(accuracy.max(DEFAULT_ACCURACY / k))
+ }
+
+ val kryoSerializer: KryoSerializer = new KryoSerializer(new SparkConf())
+
+ import collection.JavaConverters._
+
+ class NGramBuffer(val n: Int,
+ val k: Int,
+ val precisionFactor: Int,
+ val frequencyMap: HashMap[Vector[UTF8String], Double]) {
+ def add(ng: Vector[UTF8String]): Unit = {
+ var currentFrequency: Double = frequencyMap.get(ng)
+ if (currentFrequency == null.asInstanceOf[Double]) {
+ currentFrequency = 1.0D
+ } else {
+ currentFrequency += 1
+ }
+ frequencyMap.put(ng, currentFrequency)
+ }
+
+ def merge(other: NGramBuffer): Unit = {
+ other.frequencyMap.asScala.foreach((keyValuePair: (Vector[UTF8String], Double)) => {
+ val key = keyValuePair._1
+ val value = keyValuePair._2
+ val originalValue = frequencyMap.getOrDefault(key, 0.0D)
+ frequencyMap.put(key, originalValue + value)
+ })
+ }
+
+ def sortWithTwoFields(frequencyDescend: Boolean)
+ (keyWithFrequency: Tuple2[Vector[UTF8String], Double],
+ keyWithFrequency2: Tuple2[Vector[UTF8String], Double]): Boolean = {
+ if (keyWithFrequency._2 != keyWithFrequency2._2) {
+ (keyWithFrequency._2 < keyWithFrequency2._2) ^ frequencyDescend
--- End diff --
wrong ident.
---
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 #17359: [SPARK-20028][SQL] Add aggreagate expression nGra...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/17359#discussion_r107353157
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/NGrams.scala ---
@@ -0,0 +1,258 @@
+/*
+ * 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 java.util.HashMap
+
+import org.apache.spark.serializer.KryoSerializer
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult._
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription, ImplicitCastInputTypes, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.NGrams.NGramBuffer
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData}
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Return the top-k n-grams in rows that consist of sequences of strings.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(expr, n, k, accuracy) - Estimates the top-k n-grams in rows that consist of sequences
+ of strings, represented as arrays of strings, or arrays of arrays of strings. 'accuracy' is an
+ optional precision factor that controls memory usage.
+ The parameter 'n' specifies what type of n-grams are being estimated. Unigrams are n = 1, and
+ bigrams are n = 2. Generally, n will not be greater than about 5. The 'k' parameter specifies
+ how many of the highest-frequency n-grams will be returned by the UDAF. The optional precision
+ factor 'accuracy' specifies how much memory to use for estimation; more memory will give
+ more accurate frequency counts, but could crash the JVM. The value will be the max between
+ 'accuracy'(0 if it's not specified) and 1000/k, which indicates the max number of n-grams
+ which are kept in the internal HashMap.
+ The output is an array of maps with the top-k n-grams and corresponding frequency.
+ """,
+ extended = """
+ Examples:
+ > SELECT ngrams(array("abc", "abc", "bcd", "abc", "bcd"), 2, 4);
+ [{["abc","bcd"]:2.0},
+ {["abc","abc"]:1.0},
+ {["bcd","abc"]:1.0}]
+ """)
+case class NGrams(
+ child: Expression,
+ nExpression: Expression,
+ kExpression: Expression,
+ accuracyExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int)
+ extends TypedImperativeAggregate[NGramBuffer] with ImplicitCastInputTypes {
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression,
+ accuracyExpression: Expression) = {
+ this(child, nExpression, kExpression, NGrams.getAccuracy(kExpression, accuracyExpression), 0, 0)
+ }
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression) = {
+ this(child, nExpression, kExpression, Literal(0))
+ }
+
+ private lazy val n: Int = nExpression.eval().asInstanceOf[Int]
+ private lazy val k: Int = kExpression.eval().asInstanceOf[Int]
+ private lazy val accuracy: Int = accuracyExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(ArrayType(StringType, false), ArrayType(ArrayType(StringType, false))),
+ IntegerType, IntegerType, IntegerType)
+ }
+
+ val isArrayOfString = child.dataType == ArrayType(StringType, false) ||
+ child.dataType == ArrayType(StringType, true)
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (!nExpression.foldable || !kExpression.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 {
+ TypeCheckSuccess
+ }
+ }
+
+ override def createAggregationBuffer(): NGramBuffer = {
+ new NGramBuffer(n, k, accuracy, new HashMap[Vector[UTF8String], Double]())
+ }
+
+ def updateArray(genericArrayData: GenericArrayData, buffer: NGramBuffer, inputRow: InternalRow) {
+ val values = (0 until genericArrayData.numElements()).map(genericArrayData.get(_, StringType).
+ asInstanceOf[UTF8String]).toVector
+ val nGrams = getNGrams(values, n)
+ nGrams.foreach(buffer.add(_))
+ buffer.trim()
+ }
+ override def update(buffer: NGramBuffer, inputRow: InternalRow): NGramBuffer = {
+ if (isArrayOfString) {
+ updateArray(child.eval(inputRow).asInstanceOf[GenericArrayData], buffer, inputRow)
+ }
+ else {
+ val arrayOfArray = child.eval(inputRow).asInstanceOf[GenericArrayData]
+ for (i <- 0 until arrayOfArray.numElements()) {
+ updateArray(arrayOfArray.getArray(i).asInstanceOf[GenericArrayData], buffer, inputRow)
+ }
+ }
+ buffer
+ }
+
+ override def merge(buffer: NGramBuffer, input: NGramBuffer): NGramBuffer = {
+ buffer.merge(input)
+ buffer.trim()
+ buffer
+ }
+
+ override def eval(buffer: NGramBuffer): Any = {
+ val topKNGrams = buffer.getTopKNGrams().map((keyValuePair: (Vector[UTF8String], Double)) => {
+ val arrayKey = new GenericArrayData(Vector(new GenericArrayData(keyValuePair._1)))
+ val arrayValue = new GenericArrayData(Vector(keyValuePair._2))
+ new ArrayBasedMapData(arrayKey, arrayValue)
+ }).toVector
+ new GenericArrayData(topKNGrams)
+ }
+
+ private def getNGrams(values: Vector[UTF8String], n: Int): Vector[Vector[UTF8String]] = {
+ if (values.length >= n) {
+ values.sliding(n).toVector
+ }
+ else {
+ Vector()
+ }
+ }
+
+ override def withNewMutableAggBufferOffset(newOffset: Int): NGrams =
+ copy(mutableAggBufferOffset = newOffset)
+
+ override def withNewInputAggBufferOffset(newOffset: Int): NGrams =
+ copy(inputAggBufferOffset = newOffset)
+
+ override def children: Seq[Expression] = Seq(child, nExpression, kExpression, accuracyExpression)
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = ArrayType(MapType(ArrayType(StringType), DoubleType))
+
+ override def prettyName: String = "ngrams"
+
+ override def serialize(obj: NGramBuffer): Array[Byte] = {
+ NGrams.serializer.serialize(obj)
+ }
+
+ override def deserialize(bytes: Array[Byte]): NGramBuffer = {
+ NGrams.serializer.deserialize(bytes)
+ }
+}
+
+object NGrams {
+
+ val DEFAULT_ACCURACY: Int = 1000
+
+ private def getAccuracy(kExpression: Expression, accuracyExpression: Expression): Expression = {
+ val accuracy = accuracyExpression.eval().asInstanceOf[Int]
+ val k = kExpression.eval().asInstanceOf[Int]
+ Literal(accuracy.max(DEFAULT_ACCURACY / k))
+ }
+
+ val kryoSerializer: KryoSerializer = new KryoSerializer(new SparkConf())
+
+ import collection.JavaConverters._
+
+ class NGramBuffer(val n: Int,
+ val k: Int,
+ val precisionFactor: Int,
+ val frequencyMap: HashMap[Vector[UTF8String], Double]) {
+ def add(ng: Vector[UTF8String]): Unit = {
+ var currentFrequency: Double = frequencyMap.get(ng)
+ if (currentFrequency == null.asInstanceOf[Double]) {
+ currentFrequency = 1.0D
+ } else {
+ currentFrequency += 1
+ }
+ frequencyMap.put(ng, currentFrequency)
+ }
+
+ def merge(other: NGramBuffer): Unit = {
+ other.frequencyMap.asScala.foreach((keyValuePair: (Vector[UTF8String], Double)) => {
+ val key = keyValuePair._1
+ val value = keyValuePair._2
+ val originalValue = frequencyMap.getOrDefault(key, 0.0D)
+ frequencyMap.put(key, originalValue + value)
+ })
+ }
+
+ def sortWithTwoFields(frequencyDescend: Boolean)
+ (keyWithFrequency: Tuple2[Vector[UTF8String], Double],
+ keyWithFrequency2: Tuple2[Vector[UTF8String], Double]): Boolean = {
+ if (keyWithFrequency._2 != keyWithFrequency2._2) {
+ (keyWithFrequency._2 < keyWithFrequency2._2) ^ frequencyDescend
+ }
+ else {
+ val keyVector = keyWithFrequency._1
+ val keyVector2 = keyWithFrequency2._1
+ for (i <- 0 until keyVector.length) {
+ if (keyVector(i) != keyVector2(i)) {
+ return (keyVector(i).compare(keyVector2(i))) < 0
+ }
+ }
+ true
+ }
+ }
+
+ def trim(): Unit = {
+ if (frequencyMap.size() > 2 * k * precisionFactor) {
+ val orderedWithIndex = frequencyMap.asScala.iterator.toVector.
+ sortWith(sortWithTwoFields(frequencyDescend = false)).zipWithIndex
+ orderedWithIndex.takeWhile(_._2 < frequencyMap.size() - k * precisionFactor).map(_._1).
--- End diff --
frequencyMap.asScala.iterator.toVector.sortWith(sortWithTwoFields(frequencyDescend = false)).take(frequencyMap.size() - k * precisionFactor).foreach(...)
---
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 #17359: [SPARK-20028][SQL] Add aggreagate expression nGra...
Posted by gczsjdy <gi...@git.apache.org>.
Github user gczsjdy closed the pull request at:
https://github.com/apache/spark/pull/17359
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #17359: [SPARK-20028][SQL] Add aggreagate expression nGra...
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/17359#discussion_r110568613
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/NGrams.scala ---
@@ -0,0 +1,258 @@
+/*
+ * 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 java.util.HashMap
+
+import org.apache.spark.serializer.KryoSerializer
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult._
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription, ImplicitCastInputTypes, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.NGrams.NGramBuffer
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData}
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Return the top-k n-grams in rows that consist of sequences of strings.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(expr, n, k, accuracy) - Estimates the top-k n-grams in rows that consist of sequences
+ of strings, represented as arrays of strings, or arrays of arrays of strings. 'accuracy' is an
+ optional precision factor that controls memory usage.
+ The parameter 'n' specifies what type of n-grams are being estimated. Unigrams are n = 1, and
+ bigrams are n = 2. Generally, n will not be greater than about 5. The 'k' parameter specifies
+ how many of the highest-frequency n-grams will be returned by the UDAF. The optional precision
+ factor 'accuracy' specifies how much memory to use for estimation; more memory will give
+ more accurate frequency counts, but could crash the JVM. The value will be the max between
+ 'accuracy'(0 if it's not specified) and 1000/k, which indicates the max number of n-grams
+ which are kept in the internal HashMap.
+ The output is an array of maps with the top-k n-grams and corresponding frequency.
+ """,
+ extended = """
+ Examples:
+ > SELECT ngrams(array("abc", "abc", "bcd", "abc", "bcd"), 2, 4);
+ [{["abc","bcd"]:2.0},
+ {["abc","abc"]:1.0},
+ {["bcd","abc"]:1.0}]
+ """)
+case class NGrams(
+ child: Expression,
+ nExpression: Expression,
+ kExpression: Expression,
+ accuracyExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int)
+ extends TypedImperativeAggregate[NGramBuffer] with ImplicitCastInputTypes {
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression,
+ accuracyExpression: Expression) = {
+ this(child, nExpression, kExpression, NGrams.getAccuracy(kExpression, accuracyExpression), 0, 0)
+ }
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression) = {
+ this(child, nExpression, kExpression, Literal(0))
+ }
+
+ private lazy val n: Int = nExpression.eval().asInstanceOf[Int]
+ private lazy val k: Int = kExpression.eval().asInstanceOf[Int]
+ private lazy val accuracy: Int = accuracyExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(ArrayType(StringType, false), ArrayType(ArrayType(StringType, false))),
+ IntegerType, IntegerType, IntegerType)
+ }
+
+ val isArrayOfString = child.dataType == ArrayType(StringType, false) ||
+ child.dataType == ArrayType(StringType, true)
--- End diff --
what's the behavior of hive? The `ArrayType.containsNull` is just a hint, there may be no null values even the `containsNull` is true.
---
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 #17359: [SPARK-20028][SQL] Add aggreagate expression nGra...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/17359#discussion_r107348281
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/NGrams.scala ---
@@ -0,0 +1,258 @@
+/*
+ * 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 java.util.HashMap
+
+import org.apache.spark.serializer.KryoSerializer
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult._
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription, ImplicitCastInputTypes, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.NGrams.NGramBuffer
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData}
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Return the top-k n-grams in rows that consist of sequences of strings.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(expr, n, k, accuracy) - Estimates the top-k n-grams in rows that consist of sequences
+ of strings, represented as arrays of strings, or arrays of arrays of strings. 'accuracy' is an
+ optional precision factor that controls memory usage.
+ The parameter 'n' specifies what type of n-grams are being estimated. Unigrams are n = 1, and
+ bigrams are n = 2. Generally, n will not be greater than about 5. The 'k' parameter specifies
+ how many of the highest-frequency n-grams will be returned by the UDAF. The optional precision
+ factor 'accuracy' specifies how much memory to use for estimation; more memory will give
+ more accurate frequency counts, but could crash the JVM. The value will be the max between
+ 'accuracy'(0 if it's not specified) and 1000/k, which indicates the max number of n-grams
+ which are kept in the internal HashMap.
+ The output is an array of maps with the top-k n-grams and corresponding frequency.
+ """,
+ extended = """
+ Examples:
+ > SELECT ngrams(array("abc", "abc", "bcd", "abc", "bcd"), 2, 4);
+ [{["abc","bcd"]:2.0},
+ {["abc","abc"]:1.0},
+ {["bcd","abc"]:1.0}]
+ """)
+case class NGrams(
+ child: Expression,
+ nExpression: Expression,
+ kExpression: Expression,
+ accuracyExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int)
+ extends TypedImperativeAggregate[NGramBuffer] with ImplicitCastInputTypes {
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression,
+ accuracyExpression: Expression) = {
+ this(child, nExpression, kExpression, NGrams.getAccuracy(kExpression, accuracyExpression), 0, 0)
+ }
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression) = {
+ this(child, nExpression, kExpression, Literal(0))
+ }
+
+ private lazy val n: Int = nExpression.eval().asInstanceOf[Int]
+ private lazy val k: Int = kExpression.eval().asInstanceOf[Int]
+ private lazy val accuracy: Int = accuracyExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(ArrayType(StringType, false), ArrayType(ArrayType(StringType, false))),
+ IntegerType, IntegerType, IntegerType)
+ }
+
+ val isArrayOfString = child.dataType == ArrayType(StringType, false) ||
+ child.dataType == ArrayType(StringType, true)
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (!nExpression.foldable || !kExpression.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 {
+ TypeCheckSuccess
+ }
+ }
+
+ override def createAggregationBuffer(): NGramBuffer = {
+ new NGramBuffer(n, k, accuracy, new HashMap[Vector[UTF8String], Double]())
+ }
+
+ def updateArray(genericArrayData: GenericArrayData, buffer: NGramBuffer, inputRow: InternalRow) {
+ val values = (0 until genericArrayData.numElements()).map(genericArrayData.get(_, StringType).
+ asInstanceOf[UTF8String]).toVector
+ val nGrams = getNGrams(values, n)
+ nGrams.foreach(buffer.add(_))
+ buffer.trim()
+ }
+ override def update(buffer: NGramBuffer, inputRow: InternalRow): NGramBuffer = {
+ if (isArrayOfString) {
+ updateArray(child.eval(inputRow).asInstanceOf[GenericArrayData], buffer, inputRow)
+ }
+ else {
+ val arrayOfArray = child.eval(inputRow).asInstanceOf[GenericArrayData]
+ for (i <- 0 until arrayOfArray.numElements()) {
+ updateArray(arrayOfArray.getArray(i).asInstanceOf[GenericArrayData], buffer, inputRow)
+ }
+ }
+ buffer
+ }
+
+ override def merge(buffer: NGramBuffer, input: NGramBuffer): NGramBuffer = {
+ buffer.merge(input)
+ buffer.trim()
+ buffer
+ }
+
+ override def eval(buffer: NGramBuffer): Any = {
+ val topKNGrams = buffer.getTopKNGrams().map((keyValuePair: (Vector[UTF8String], Double)) => {
+ val arrayKey = new GenericArrayData(Vector(new GenericArrayData(keyValuePair._1)))
+ val arrayValue = new GenericArrayData(Vector(keyValuePair._2))
+ new ArrayBasedMapData(arrayKey, arrayValue)
+ }).toVector
+ new GenericArrayData(topKNGrams)
+ }
+
+ private def getNGrams(values: Vector[UTF8String], n: Int): Vector[Vector[UTF8String]] = {
+ if (values.length >= n) {
+ values.sliding(n).toVector
+ }
+ else {
+ Vector()
+ }
+ }
+
+ override def withNewMutableAggBufferOffset(newOffset: Int): NGrams =
+ copy(mutableAggBufferOffset = newOffset)
+
+ override def withNewInputAggBufferOffset(newOffset: Int): NGrams =
+ copy(inputAggBufferOffset = newOffset)
+
+ override def children: Seq[Expression] = Seq(child, nExpression, kExpression, accuracyExpression)
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = ArrayType(MapType(ArrayType(StringType), DoubleType))
+
+ override def prettyName: String = "ngrams"
+
+ override def serialize(obj: NGramBuffer): Array[Byte] = {
+ NGrams.serializer.serialize(obj)
+ }
+
+ override def deserialize(bytes: Array[Byte]): NGramBuffer = {
+ NGrams.serializer.deserialize(bytes)
+ }
+}
+
+object NGrams {
+
+ val DEFAULT_ACCURACY: Int = 1000
+
+ private def getAccuracy(kExpression: Expression, accuracyExpression: Expression): Expression = {
+ val accuracy = accuracyExpression.eval().asInstanceOf[Int]
+ val k = kExpression.eval().asInstanceOf[Int]
+ Literal(accuracy.max(DEFAULT_ACCURACY / k))
+ }
+
+ val kryoSerializer: KryoSerializer = new KryoSerializer(new SparkConf())
+
+ import collection.JavaConverters._
+
+ class NGramBuffer(val n: Int,
+ val k: Int,
+ val precisionFactor: Int,
+ val frequencyMap: HashMap[Vector[UTF8String], Double]) {
+ def add(ng: Vector[UTF8String]): Unit = {
+ var currentFrequency: Double = frequencyMap.get(ng)
+ if (currentFrequency == null.asInstanceOf[Double]) {
--- End diff --
frequencyMap.put(ng, frequencyMap.getOrElse(ng, 0.0D) + 1)
---
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 #17359: [SPARK-20028][SQL] Add aggreagate expression nGra...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/17359#discussion_r107348955
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/NGrams.scala ---
@@ -0,0 +1,258 @@
+/*
+ * 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 java.util.HashMap
+
+import org.apache.spark.serializer.KryoSerializer
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult._
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription, ImplicitCastInputTypes, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.NGrams.NGramBuffer
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData}
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Return the top-k n-grams in rows that consist of sequences of strings.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(expr, n, k, accuracy) - Estimates the top-k n-grams in rows that consist of sequences
+ of strings, represented as arrays of strings, or arrays of arrays of strings. 'accuracy' is an
+ optional precision factor that controls memory usage.
+ The parameter 'n' specifies what type of n-grams are being estimated. Unigrams are n = 1, and
+ bigrams are n = 2. Generally, n will not be greater than about 5. The 'k' parameter specifies
+ how many of the highest-frequency n-grams will be returned by the UDAF. The optional precision
+ factor 'accuracy' specifies how much memory to use for estimation; more memory will give
+ more accurate frequency counts, but could crash the JVM. The value will be the max between
+ 'accuracy'(0 if it's not specified) and 1000/k, which indicates the max number of n-grams
+ which are kept in the internal HashMap.
+ The output is an array of maps with the top-k n-grams and corresponding frequency.
+ """,
+ extended = """
+ Examples:
+ > SELECT ngrams(array("abc", "abc", "bcd", "abc", "bcd"), 2, 4);
+ [{["abc","bcd"]:2.0},
+ {["abc","abc"]:1.0},
+ {["bcd","abc"]:1.0}]
+ """)
+case class NGrams(
+ child: Expression,
+ nExpression: Expression,
+ kExpression: Expression,
+ accuracyExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int)
+ extends TypedImperativeAggregate[NGramBuffer] with ImplicitCastInputTypes {
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression,
+ accuracyExpression: Expression) = {
+ this(child, nExpression, kExpression, NGrams.getAccuracy(kExpression, accuracyExpression), 0, 0)
+ }
+
+ def this(child: Expression, nExpression: Expression, kExpression: Expression) = {
+ this(child, nExpression, kExpression, Literal(0))
+ }
+
+ private lazy val n: Int = nExpression.eval().asInstanceOf[Int]
+ private lazy val k: Int = kExpression.eval().asInstanceOf[Int]
+ private lazy val accuracy: Int = accuracyExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(ArrayType(StringType, false), ArrayType(ArrayType(StringType, false))),
+ IntegerType, IntegerType, IntegerType)
+ }
+
+ val isArrayOfString = child.dataType == ArrayType(StringType, false) ||
+ child.dataType == ArrayType(StringType, true)
--- End diff --
nvm...
---
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