You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2022/10/24 10:39:07 UTC
[GitHub] [spark] zhengruifeng opened a new pull request, #38375: [SPARK-40900][SQL] Reimplement `frequentItems` with dataframe operations
zhengruifeng opened a new pull request, #38375:
URL: https://github.com/apache/spark/pull/38375
### What changes were proposed in this pull request?
Reimplement `frequentItems` with dataframe operations
### Why are the changes needed?
1, do not truncate the sql plan any more;
2, enable sql optimization like column pruning
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
existing UTs and manually check
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] zhengruifeng commented on a diff in pull request #38375: [SPARK-40900][SQL] Reimplement `frequentItems` with dataframe operations
Posted by GitBox <gi...@apache.org>.
zhengruifeng commented on code in PR #38375:
URL: https://github.com/apache/spark/pull/38375#discussion_r1004099819
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/stat/FrequentItems.scala:
##########
@@ -85,42 +51,142 @@ object FrequentItems extends Logging {
cols: Seq[String],
support: Double): DataFrame = {
require(support >= 1e-4 && support <= 1.0, s"Support must be in [1e-4, 1], but got $support.")
- val numCols = cols.length
+
// number of max items to keep counts for
val sizeOfMap = (1 / support).toInt
- val countMaps = Seq.tabulate(numCols)(i => new FreqItemCounter(sizeOfMap))
-
- val freqItems = df.select(cols.map(Column(_)) : _*).rdd.treeAggregate(countMaps)(
- seqOp = (counts, row) => {
- var i = 0
- while (i < numCols) {
- val thisMap = counts(i)
- val key = row.get(i)
- thisMap.add(key, 1L)
- i += 1
- }
- counts
- },
- combOp = (baseCounts, counts) => {
- var i = 0
- while (i < numCols) {
- baseCounts(i).merge(counts(i))
- i += 1
+
+ val frequentItemCols = cols.map { col =>
+ val aggExpr = new CollectFrequentItems(functions.col(col).expr, sizeOfMap)
+ Column(aggExpr.toAggregateExpression(isDistinct = false)).as(s"${col}_freqItems")
+ }
+
+ df.select(frequentItemCols: _*)
+ }
+}
+
+case class CollectFrequentItems(
+ child: Expression,
+ size: Int,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends TypedImperativeAggregate[mutable.Map[Any, Long]]
+ with ImplicitCastInputTypes with UnaryLike[Expression] {
+ require(size > 0)
+
+ def this(child: Expression, size: Int) = this(child, size, 0, 0)
+
+ // Returns empty array for empty inputs
+ override def nullable: Boolean = false
+
+ override def dataType: DataType = ArrayType(child.dataType, containsNull = child.nullable)
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(AnyDataType)
+
+ override def prettyName: String = "collect_frequent_items"
+
+ override def createAggregationBuffer(): mutable.Map[Any, Long] =
+ mutable.Map.empty[Any, Long]
+
+ private def add(map: mutable.Map[Any, Long], key: Any, count: Long): mutable.Map[Any, Long] = {
+ if (map.contains(key)) {
+ map(key) += count
+ } else {
+ if (map.size < size) {
+ map += key -> count
+ } else {
+ val minCount = if (map.values.isEmpty) 0 else map.values.min
+ val remainder = count - minCount
+ if (remainder >= 0) {
+ map += key -> count // something will get kicked out, so we can add this
+ map.retain((k, v) => v > minCount)
+ map.transform((k, v) => v - minCount)
+ } else {
+ map.transform((k, v) => v - count)
}
- baseCounts
}
- )
- val justItems = freqItems.map(m => m.baseMap.keys.toArray)
- val resultRow = Row(justItems : _*)
+ }
+ map
+ }
+
+ override def update(
+ buffer: mutable.Map[Any, Long],
+ input: InternalRow): mutable.Map[Any, Long] = {
+ val key = child.eval(input)
+ if (key != null) {
+ this.add(buffer, InternalRow.copyValue(key), 1L)
+ } else {
+ this.add(buffer, key, 1L)
+ }
+ }
+
+ override def merge(
+ buffer: mutable.Map[Any, Long],
+ input: mutable.Map[Any, Long]): mutable.Map[Any, Long] = {
+ input.foreach { case (k, v) =>
+ add(buffer, k, v)
+ }
+ buffer
+ }
- val outputCols = cols.map { name =>
- val originalField = df.resolve(name)
+ override def eval(buffer: mutable.Map[Any, Long]): Any =
+ new GenericArrayData(buffer.keys.toArray)
- // append frequent Items to the column name for easy debugging
- StructField(name + "_freqItems", ArrayType(originalField.dataType, originalField.nullable))
- }.toArray
+ private lazy val projection =
+ UnsafeProjection.create(Array[DataType](child.dataType, LongType))
- val schema = StructType(outputCols).toAttributes
- Dataset.ofRows(df.sparkSession, LocalRelation.fromExternalRows(schema, Seq(resultRow)))
+ override def serialize(map: mutable.Map[Any, Long]): Array[Byte] = {
+ val buffer = new Array[Byte](4 << 10) // 4K
+ val bos = new ByteArrayOutputStream()
+ val out = new DataOutputStream(bos)
+ try {
Review Comment:
cool, let me update it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on pull request #38375: [SPARK-40900][SQL] Reimplement `frequentItems` with dataframe operations
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on PR #38375:
URL: https://github.com/apache/spark/pull/38375#issuecomment-1288841752
cc @cloud-fan FYI
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on pull request #38375: [SPARK-40900][SQL] Reimplement `frequentItems` with dataframe operations
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on PR #38375:
URL: https://github.com/apache/spark/pull/38375#issuecomment-1288830300
this is nice!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on a diff in pull request #38375: [SPARK-40900][SQL] Reimplement `frequentItems` with dataframe operations
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on code in PR #38375:
URL: https://github.com/apache/spark/pull/38375#discussion_r1004085893
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/stat/FrequentItems.scala:
##########
@@ -85,42 +51,142 @@ object FrequentItems extends Logging {
cols: Seq[String],
support: Double): DataFrame = {
require(support >= 1e-4 && support <= 1.0, s"Support must be in [1e-4, 1], but got $support.")
- val numCols = cols.length
+
// number of max items to keep counts for
val sizeOfMap = (1 / support).toInt
- val countMaps = Seq.tabulate(numCols)(i => new FreqItemCounter(sizeOfMap))
-
- val freqItems = df.select(cols.map(Column(_)) : _*).rdd.treeAggregate(countMaps)(
- seqOp = (counts, row) => {
- var i = 0
- while (i < numCols) {
- val thisMap = counts(i)
- val key = row.get(i)
- thisMap.add(key, 1L)
- i += 1
- }
- counts
- },
- combOp = (baseCounts, counts) => {
- var i = 0
- while (i < numCols) {
- baseCounts(i).merge(counts(i))
- i += 1
+
+ val frequentItemCols = cols.map { col =>
+ val aggExpr = new CollectFrequentItems(functions.col(col).expr, sizeOfMap)
+ Column(aggExpr.toAggregateExpression(isDistinct = false)).as(s"${col}_freqItems")
+ }
+
+ df.select(frequentItemCols: _*)
+ }
+}
+
+case class CollectFrequentItems(
+ child: Expression,
+ size: Int,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends TypedImperativeAggregate[mutable.Map[Any, Long]]
+ with ImplicitCastInputTypes with UnaryLike[Expression] {
+ require(size > 0)
+
+ def this(child: Expression, size: Int) = this(child, size, 0, 0)
+
+ // Returns empty array for empty inputs
+ override def nullable: Boolean = false
+
+ override def dataType: DataType = ArrayType(child.dataType, containsNull = child.nullable)
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(AnyDataType)
+
+ override def prettyName: String = "collect_frequent_items"
+
+ override def createAggregationBuffer(): mutable.Map[Any, Long] =
+ mutable.Map.empty[Any, Long]
+
+ private def add(map: mutable.Map[Any, Long], key: Any, count: Long): mutable.Map[Any, Long] = {
+ if (map.contains(key)) {
+ map(key) += count
+ } else {
+ if (map.size < size) {
+ map += key -> count
+ } else {
+ val minCount = if (map.values.isEmpty) 0 else map.values.min
+ val remainder = count - minCount
+ if (remainder >= 0) {
+ map += key -> count // something will get kicked out, so we can add this
+ map.retain((k, v) => v > minCount)
+ map.transform((k, v) => v - minCount)
+ } else {
+ map.transform((k, v) => v - count)
}
- baseCounts
}
- )
- val justItems = freqItems.map(m => m.baseMap.keys.toArray)
- val resultRow = Row(justItems : _*)
+ }
+ map
+ }
+
+ override def update(
+ buffer: mutable.Map[Any, Long],
+ input: InternalRow): mutable.Map[Any, Long] = {
+ val key = child.eval(input)
+ if (key != null) {
+ this.add(buffer, InternalRow.copyValue(key), 1L)
+ } else {
+ this.add(buffer, key, 1L)
+ }
+ }
+
+ override def merge(
+ buffer: mutable.Map[Any, Long],
+ input: mutable.Map[Any, Long]): mutable.Map[Any, Long] = {
+ input.foreach { case (k, v) =>
+ add(buffer, k, v)
+ }
+ buffer
+ }
- val outputCols = cols.map { name =>
- val originalField = df.resolve(name)
+ override def eval(buffer: mutable.Map[Any, Long]): Any =
+ new GenericArrayData(buffer.keys.toArray)
- // append frequent Items to the column name for easy debugging
- StructField(name + "_freqItems", ArrayType(originalField.dataType, originalField.nullable))
- }.toArray
+ private lazy val projection =
+ UnsafeProjection.create(Array[DataType](child.dataType, LongType))
- val schema = StructType(outputCols).toAttributes
- Dataset.ofRows(df.sparkSession, LocalRelation.fromExternalRows(schema, Seq(resultRow)))
+ override def serialize(map: mutable.Map[Any, Long]): Array[Byte] = {
+ val buffer = new Array[Byte](4 << 10) // 4K
+ val bos = new ByteArrayOutputStream()
+ val out = new DataOutputStream(bos)
+ try {
Review Comment:
We could use `Utils.tryWithSafeFinally` but that's fine.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on a diff in pull request #38375: [SPARK-40900][SQL] Reimplement `frequentItems` with dataframe operations
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on code in PR #38375:
URL: https://github.com/apache/spark/pull/38375#discussion_r1003157994
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/stat/FrequentItems.scala:
##########
@@ -85,42 +51,142 @@ object FrequentItems extends Logging {
cols: Seq[String],
support: Double): DataFrame = {
require(support >= 1e-4 && support <= 1.0, s"Support must be in [1e-4, 1], but got $support.")
- val numCols = cols.length
+
// number of max items to keep counts for
val sizeOfMap = (1 / support).toInt
- val countMaps = Seq.tabulate(numCols)(i => new FreqItemCounter(sizeOfMap))
-
- val freqItems = df.select(cols.map(Column(_)) : _*).rdd.treeAggregate(countMaps)(
- seqOp = (counts, row) => {
- var i = 0
- while (i < numCols) {
- val thisMap = counts(i)
- val key = row.get(i)
- thisMap.add(key, 1L)
- i += 1
- }
- counts
- },
- combOp = (baseCounts, counts) => {
- var i = 0
- while (i < numCols) {
- baseCounts(i).merge(counts(i))
- i += 1
+
+ val frequentItemCols = cols.map { col =>
+ val aggExpr = new CollectFrequentItems(functions.col(col).expr, sizeOfMap)
+ Column(aggExpr.toAggregateExpression(isDistinct = false)).as(s"${col}_freqItems")
+ }
+
+ df.select(frequentItemCols: _*)
+ }
+}
+
+case class CollectFrequentItems(
+ child: Expression,
+ size: Int,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends TypedImperativeAggregate[mutable.Map[Any, Long]]
+ with ImplicitCastInputTypes with UnaryLike[Expression] {
+ require(size > 0)
+
+ def this(child: Expression, size: Int) = this(child, size, 0, 0)
+
+ // Returns empty array for empty inputs
+ override def nullable: Boolean = false
+
+ override def dataType: DataType = ArrayType(child.dataType, containsNull = child.nullable)
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(AnyDataType)
+
+ override def prettyName: String = "collect_frequent_items"
+
+ override def createAggregationBuffer(): mutable.Map[Any, Long] =
+ mutable.Map.empty[Any, Long]
+
+ private def add(map: mutable.Map[Any, Long], key: Any, count: Long): mutable.Map[Any, Long] = {
+ if (map.contains(key)) {
+ map(key) += count
+ } else {
+ if (map.size < size) {
+ map += key -> count
+ } else {
+ val minCount = if (map.values.isEmpty) 0 else map.values.min
+ val remainder = count - minCount
+ if (remainder >= 0) {
+ map += key -> count // something will get kicked out, so we can add this
+ map.retain((k, v) => v > minCount)
+ map.transform((k, v) => v - minCount)
+ } else {
+ map.transform((k, v) => v - count)
}
- baseCounts
}
- )
- val justItems = freqItems.map(m => m.baseMap.keys.toArray)
- val resultRow = Row(justItems : _*)
+ }
+ map
+ }
+
+ override def update(
+ buffer: mutable.Map[Any, Long],
+ input: InternalRow): mutable.Map[Any, Long] = {
+ val key = child.eval(input)
+ if (key != null) {
+ this.add(buffer, InternalRow.copyValue(key), 1L)
+ } else {
+ this.add(buffer, key, 1L)
+ }
+ }
+
+ override def merge(
+ buffer: mutable.Map[Any, Long],
+ input: mutable.Map[Any, Long]): mutable.Map[Any, Long] = {
+ input.foreach { case (k, v) =>
+ add(buffer, k, v)
+ }
+ buffer
+ }
- val outputCols = cols.map { name =>
- val originalField = df.resolve(name)
+ override def eval(buffer: mutable.Map[Any, Long]): Any =
+ new GenericArrayData(buffer.keys.toArray)
- // append frequent Items to the column name for easy debugging
- StructField(name + "_freqItems", ArrayType(originalField.dataType, originalField.nullable))
- }.toArray
+ private lazy val projection =
+ UnsafeProjection.create(Array[DataType](child.dataType, LongType))
- val schema = StructType(outputCols).toAttributes
- Dataset.ofRows(df.sparkSession, LocalRelation.fromExternalRows(schema, Seq(resultRow)))
+ override def serialize(map: mutable.Map[Any, Long]): Array[Byte] = {
+ val buffer = new Array[Byte](4 << 10) // 4K
+ val bos = new ByteArrayOutputStream()
+ val out = new DataOutputStream(bos)
+ try {
Review Comment:
You can leverage `Utils.tryWithResource`
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/stat/FrequentItems.scala:
##########
@@ -85,42 +51,142 @@ object FrequentItems extends Logging {
cols: Seq[String],
support: Double): DataFrame = {
require(support >= 1e-4 && support <= 1.0, s"Support must be in [1e-4, 1], but got $support.")
- val numCols = cols.length
+
// number of max items to keep counts for
val sizeOfMap = (1 / support).toInt
- val countMaps = Seq.tabulate(numCols)(i => new FreqItemCounter(sizeOfMap))
-
- val freqItems = df.select(cols.map(Column(_)) : _*).rdd.treeAggregate(countMaps)(
- seqOp = (counts, row) => {
- var i = 0
- while (i < numCols) {
- val thisMap = counts(i)
- val key = row.get(i)
- thisMap.add(key, 1L)
- i += 1
- }
- counts
- },
- combOp = (baseCounts, counts) => {
- var i = 0
- while (i < numCols) {
- baseCounts(i).merge(counts(i))
- i += 1
+
+ val frequentItemCols = cols.map { col =>
+ val aggExpr = new CollectFrequentItems(functions.col(col).expr, sizeOfMap)
+ Column(aggExpr.toAggregateExpression(isDistinct = false)).as(s"${col}_freqItems")
+ }
+
+ df.select(frequentItemCols: _*)
+ }
+}
+
+case class CollectFrequentItems(
+ child: Expression,
+ size: Int,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends TypedImperativeAggregate[mutable.Map[Any, Long]]
+ with ImplicitCastInputTypes with UnaryLike[Expression] {
+ require(size > 0)
+
+ def this(child: Expression, size: Int) = this(child, size, 0, 0)
+
+ // Returns empty array for empty inputs
+ override def nullable: Boolean = false
+
+ override def dataType: DataType = ArrayType(child.dataType, containsNull = child.nullable)
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(AnyDataType)
+
+ override def prettyName: String = "collect_frequent_items"
+
+ override def createAggregationBuffer(): mutable.Map[Any, Long] =
+ mutable.Map.empty[Any, Long]
+
+ private def add(map: mutable.Map[Any, Long], key: Any, count: Long): mutable.Map[Any, Long] = {
+ if (map.contains(key)) {
+ map(key) += count
+ } else {
+ if (map.size < size) {
+ map += key -> count
+ } else {
+ val minCount = if (map.values.isEmpty) 0 else map.values.min
+ val remainder = count - minCount
+ if (remainder >= 0) {
+ map += key -> count // something will get kicked out, so we can add this
+ map.retain((k, v) => v > minCount)
+ map.transform((k, v) => v - minCount)
+ } else {
+ map.transform((k, v) => v - count)
}
- baseCounts
}
- )
- val justItems = freqItems.map(m => m.baseMap.keys.toArray)
- val resultRow = Row(justItems : _*)
+ }
+ map
+ }
+
+ override def update(
+ buffer: mutable.Map[Any, Long],
+ input: InternalRow): mutable.Map[Any, Long] = {
+ val key = child.eval(input)
+ if (key != null) {
+ this.add(buffer, InternalRow.copyValue(key), 1L)
+ } else {
+ this.add(buffer, key, 1L)
+ }
+ }
+
+ override def merge(
+ buffer: mutable.Map[Any, Long],
+ input: mutable.Map[Any, Long]): mutable.Map[Any, Long] = {
+ input.foreach { case (k, v) =>
+ add(buffer, k, v)
+ }
+ buffer
+ }
- val outputCols = cols.map { name =>
- val originalField = df.resolve(name)
+ override def eval(buffer: mutable.Map[Any, Long]): Any =
+ new GenericArrayData(buffer.keys.toArray)
- // append frequent Items to the column name for easy debugging
- StructField(name + "_freqItems", ArrayType(originalField.dataType, originalField.nullable))
- }.toArray
+ private lazy val projection =
+ UnsafeProjection.create(Array[DataType](child.dataType, LongType))
- val schema = StructType(outputCols).toAttributes
- Dataset.ofRows(df.sparkSession, LocalRelation.fromExternalRows(schema, Seq(resultRow)))
+ override def serialize(map: mutable.Map[Any, Long]): Array[Byte] = {
+ val buffer = new Array[Byte](4 << 10) // 4K
+ val bos = new ByteArrayOutputStream()
+ val out = new DataOutputStream(bos)
+ try {
+ // Write pairs in counts map to byte buffer.
+ map.foreach { case (key, count) =>
+ val row = InternalRow.apply(key, count)
+ val unsafeRow = projection.apply(row)
+ out.writeInt(unsafeRow.getSizeInBytes)
+ unsafeRow.writeToStream(out, buffer)
+ }
+ out.writeInt(-1)
+ out.flush()
+
+ bos.toByteArray
+ } finally {
+ out.close()
+ bos.close()
+ }
}
+
+ override def deserialize(bytes: Array[Byte]): mutable.Map[Any, Long] = {
+ val bis = new ByteArrayInputStream(bytes)
+ val ins = new DataInputStream(bis)
+ try {
Review Comment:
ditto
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on a diff in pull request #38375: [SPARK-40900][SQL] Reimplement `frequentItems` with dataframe operations
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on code in PR #38375:
URL: https://github.com/apache/spark/pull/38375#discussion_r1003159352
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/stat/FrequentItems.scala:
##########
@@ -85,42 +51,142 @@ object FrequentItems extends Logging {
cols: Seq[String],
support: Double): DataFrame = {
require(support >= 1e-4 && support <= 1.0, s"Support must be in [1e-4, 1], but got $support.")
- val numCols = cols.length
+
// number of max items to keep counts for
val sizeOfMap = (1 / support).toInt
- val countMaps = Seq.tabulate(numCols)(i => new FreqItemCounter(sizeOfMap))
-
- val freqItems = df.select(cols.map(Column(_)) : _*).rdd.treeAggregate(countMaps)(
- seqOp = (counts, row) => {
- var i = 0
- while (i < numCols) {
- val thisMap = counts(i)
- val key = row.get(i)
- thisMap.add(key, 1L)
- i += 1
- }
- counts
- },
- combOp = (baseCounts, counts) => {
- var i = 0
- while (i < numCols) {
- baseCounts(i).merge(counts(i))
- i += 1
+
+ val frequentItemCols = cols.map { col =>
+ val aggExpr = new CollectFrequentItems(functions.col(col).expr, sizeOfMap)
+ Column(aggExpr.toAggregateExpression(isDistinct = false)).as(s"${col}_freqItems")
+ }
+
+ df.select(frequentItemCols: _*)
+ }
+}
+
+case class CollectFrequentItems(
+ child: Expression,
+ size: Int,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends TypedImperativeAggregate[mutable.Map[Any, Long]]
+ with ImplicitCastInputTypes with UnaryLike[Expression] {
+ require(size > 0)
+
+ def this(child: Expression, size: Int) = this(child, size, 0, 0)
+
+ // Returns empty array for empty inputs
+ override def nullable: Boolean = false
+
+ override def dataType: DataType = ArrayType(child.dataType, containsNull = child.nullable)
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(AnyDataType)
+
+ override def prettyName: String = "collect_frequent_items"
+
+ override def createAggregationBuffer(): mutable.Map[Any, Long] =
+ mutable.Map.empty[Any, Long]
+
+ private def add(map: mutable.Map[Any, Long], key: Any, count: Long): mutable.Map[Any, Long] = {
+ if (map.contains(key)) {
+ map(key) += count
+ } else {
+ if (map.size < size) {
+ map += key -> count
+ } else {
+ val minCount = if (map.values.isEmpty) 0 else map.values.min
+ val remainder = count - minCount
+ if (remainder >= 0) {
+ map += key -> count // something will get kicked out, so we can add this
+ map.retain((k, v) => v > minCount)
+ map.transform((k, v) => v - minCount)
+ } else {
+ map.transform((k, v) => v - count)
}
- baseCounts
}
- )
- val justItems = freqItems.map(m => m.baseMap.keys.toArray)
- val resultRow = Row(justItems : _*)
+ }
+ map
+ }
+
+ override def update(
+ buffer: mutable.Map[Any, Long],
+ input: InternalRow): mutable.Map[Any, Long] = {
+ val key = child.eval(input)
+ if (key != null) {
+ this.add(buffer, InternalRow.copyValue(key), 1L)
+ } else {
+ this.add(buffer, key, 1L)
+ }
+ }
+
+ override def merge(
+ buffer: mutable.Map[Any, Long],
+ input: mutable.Map[Any, Long]): mutable.Map[Any, Long] = {
+ input.foreach { case (k, v) =>
Review Comment:
If this is a hot path, you can just use a plan `for` loop or `while.`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] zhengruifeng commented on a diff in pull request #38375: [SPARK-40900][SQL] Reimplement `frequentItems` with dataframe operations
Posted by GitBox <gi...@apache.org>.
zhengruifeng commented on code in PR #38375:
URL: https://github.com/apache/spark/pull/38375#discussion_r1003975996
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/stat/FrequentItems.scala:
##########
@@ -85,42 +51,142 @@ object FrequentItems extends Logging {
cols: Seq[String],
support: Double): DataFrame = {
require(support >= 1e-4 && support <= 1.0, s"Support must be in [1e-4, 1], but got $support.")
- val numCols = cols.length
+
// number of max items to keep counts for
val sizeOfMap = (1 / support).toInt
- val countMaps = Seq.tabulate(numCols)(i => new FreqItemCounter(sizeOfMap))
-
- val freqItems = df.select(cols.map(Column(_)) : _*).rdd.treeAggregate(countMaps)(
- seqOp = (counts, row) => {
- var i = 0
- while (i < numCols) {
- val thisMap = counts(i)
- val key = row.get(i)
- thisMap.add(key, 1L)
- i += 1
- }
- counts
- },
- combOp = (baseCounts, counts) => {
- var i = 0
- while (i < numCols) {
- baseCounts(i).merge(counts(i))
- i += 1
+
+ val frequentItemCols = cols.map { col =>
+ val aggExpr = new CollectFrequentItems(functions.col(col).expr, sizeOfMap)
+ Column(aggExpr.toAggregateExpression(isDistinct = false)).as(s"${col}_freqItems")
+ }
+
+ df.select(frequentItemCols: _*)
+ }
+}
+
+case class CollectFrequentItems(
+ child: Expression,
+ size: Int,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends TypedImperativeAggregate[mutable.Map[Any, Long]]
+ with ImplicitCastInputTypes with UnaryLike[Expression] {
+ require(size > 0)
+
+ def this(child: Expression, size: Int) = this(child, size, 0, 0)
+
+ // Returns empty array for empty inputs
+ override def nullable: Boolean = false
+
+ override def dataType: DataType = ArrayType(child.dataType, containsNull = child.nullable)
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(AnyDataType)
+
+ override def prettyName: String = "collect_frequent_items"
+
+ override def createAggregationBuffer(): mutable.Map[Any, Long] =
+ mutable.Map.empty[Any, Long]
+
+ private def add(map: mutable.Map[Any, Long], key: Any, count: Long): mutable.Map[Any, Long] = {
+ if (map.contains(key)) {
+ map(key) += count
+ } else {
+ if (map.size < size) {
+ map += key -> count
+ } else {
+ val minCount = if (map.values.isEmpty) 0 else map.values.min
+ val remainder = count - minCount
+ if (remainder >= 0) {
+ map += key -> count // something will get kicked out, so we can add this
+ map.retain((k, v) => v > minCount)
+ map.transform((k, v) => v - minCount)
+ } else {
+ map.transform((k, v) => v - count)
}
- baseCounts
}
- )
- val justItems = freqItems.map(m => m.baseMap.keys.toArray)
- val resultRow = Row(justItems : _*)
+ }
+ map
+ }
+
+ override def update(
+ buffer: mutable.Map[Any, Long],
+ input: InternalRow): mutable.Map[Any, Long] = {
+ val key = child.eval(input)
+ if (key != null) {
+ this.add(buffer, InternalRow.copyValue(key), 1L)
+ } else {
+ this.add(buffer, key, 1L)
+ }
+ }
+
+ override def merge(
+ buffer: mutable.Map[Any, Long],
+ input: mutable.Map[Any, Long]): mutable.Map[Any, Long] = {
+ input.foreach { case (k, v) =>
Review Comment:
got it , will update
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] zhengruifeng commented on pull request #38375: [SPARK-40900][SQL] Reimplement `frequentItems` with dataframe operations
Posted by GitBox <gi...@apache.org>.
zhengruifeng commented on PR #38375:
URL: https://github.com/apache/spark/pull/38375#issuecomment-1288816657
cc @HyukjinKwon
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on a diff in pull request #38375: [SPARK-40900][SQL] Reimplement `frequentItems` with dataframe operations
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on code in PR #38375:
URL: https://github.com/apache/spark/pull/38375#discussion_r1003168887
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/stat/FrequentItems.scala:
##########
@@ -85,42 +51,142 @@ object FrequentItems extends Logging {
cols: Seq[String],
support: Double): DataFrame = {
require(support >= 1e-4 && support <= 1.0, s"Support must be in [1e-4, 1], but got $support.")
- val numCols = cols.length
+
// number of max items to keep counts for
val sizeOfMap = (1 / support).toInt
- val countMaps = Seq.tabulate(numCols)(i => new FreqItemCounter(sizeOfMap))
-
- val freqItems = df.select(cols.map(Column(_)) : _*).rdd.treeAggregate(countMaps)(
- seqOp = (counts, row) => {
- var i = 0
- while (i < numCols) {
- val thisMap = counts(i)
- val key = row.get(i)
- thisMap.add(key, 1L)
- i += 1
- }
- counts
- },
- combOp = (baseCounts, counts) => {
- var i = 0
- while (i < numCols) {
- baseCounts(i).merge(counts(i))
- i += 1
+
+ val frequentItemCols = cols.map { col =>
+ val aggExpr = new CollectFrequentItems(functions.col(col).expr, sizeOfMap)
+ Column(aggExpr.toAggregateExpression(isDistinct = false)).as(s"${col}_freqItems")
+ }
+
+ df.select(frequentItemCols: _*)
+ }
+}
+
+case class CollectFrequentItems(
+ child: Expression,
+ size: Int,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends TypedImperativeAggregate[mutable.Map[Any, Long]]
+ with ImplicitCastInputTypes with UnaryLike[Expression] {
+ require(size > 0)
+
+ def this(child: Expression, size: Int) = this(child, size, 0, 0)
+
+ // Returns empty array for empty inputs
+ override def nullable: Boolean = false
+
+ override def dataType: DataType = ArrayType(child.dataType, containsNull = child.nullable)
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(AnyDataType)
+
+ override def prettyName: String = "collect_frequent_items"
+
+ override def createAggregationBuffer(): mutable.Map[Any, Long] =
+ mutable.Map.empty[Any, Long]
+
+ private def add(map: mutable.Map[Any, Long], key: Any, count: Long): mutable.Map[Any, Long] = {
+ if (map.contains(key)) {
+ map(key) += count
+ } else {
+ if (map.size < size) {
+ map += key -> count
+ } else {
+ val minCount = if (map.values.isEmpty) 0 else map.values.min
+ val remainder = count - minCount
+ if (remainder >= 0) {
+ map += key -> count // something will get kicked out, so we can add this
+ map.retain((k, v) => v > minCount)
+ map.transform((k, v) => v - minCount)
+ } else {
+ map.transform((k, v) => v - count)
}
- baseCounts
}
- )
- val justItems = freqItems.map(m => m.baseMap.keys.toArray)
- val resultRow = Row(justItems : _*)
+ }
+ map
+ }
+
+ override def update(
+ buffer: mutable.Map[Any, Long],
+ input: InternalRow): mutable.Map[Any, Long] = {
+ val key = child.eval(input)
+ if (key != null) {
+ this.add(buffer, InternalRow.copyValue(key), 1L)
+ } else {
+ this.add(buffer, key, 1L)
+ }
+ }
+
+ override def merge(
+ buffer: mutable.Map[Any, Long],
+ input: mutable.Map[Any, Long]): mutable.Map[Any, Long] = {
+ input.foreach { case (k, v) =>
+ add(buffer, k, v)
+ }
+ buffer
+ }
- val outputCols = cols.map { name =>
- val originalField = df.resolve(name)
+ override def eval(buffer: mutable.Map[Any, Long]): Any =
+ new GenericArrayData(buffer.keys.toArray)
Review Comment:
Can we reuse one `GenericArrayData` with cleaning up?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a diff in pull request #38375: [SPARK-40900][SQL] Reimplement `frequentItems` with dataframe operations
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #38375:
URL: https://github.com/apache/spark/pull/38375#discussion_r1004000396
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/stat/FrequentItems.scala:
##########
@@ -85,42 +51,144 @@ object FrequentItems extends Logging {
cols: Seq[String],
support: Double): DataFrame = {
require(support >= 1e-4 && support <= 1.0, s"Support must be in [1e-4, 1], but got $support.")
- val numCols = cols.length
+
// number of max items to keep counts for
val sizeOfMap = (1 / support).toInt
- val countMaps = Seq.tabulate(numCols)(i => new FreqItemCounter(sizeOfMap))
-
- val freqItems = df.select(cols.map(Column(_)) : _*).rdd.treeAggregate(countMaps)(
- seqOp = (counts, row) => {
- var i = 0
- while (i < numCols) {
- val thisMap = counts(i)
- val key = row.get(i)
- thisMap.add(key, 1L)
- i += 1
- }
- counts
- },
- combOp = (baseCounts, counts) => {
- var i = 0
- while (i < numCols) {
- baseCounts(i).merge(counts(i))
- i += 1
+
+ val frequentItemCols = cols.map { col =>
+ val aggExpr = new CollectFrequentItems(functions.col(col).expr, sizeOfMap)
+ Column(aggExpr.toAggregateExpression(isDistinct = false)).as(s"${col}_freqItems")
+ }
+
+ df.select(frequentItemCols: _*)
+ }
+}
+
+case class CollectFrequentItems(
+ child: Expression,
+ size: Int,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends TypedImperativeAggregate[mutable.Map[Any, Long]]
+ with ImplicitCastInputTypes with UnaryLike[Expression] {
+ require(size > 0)
+
+ def this(child: Expression, size: Int) = this(child, size, 0, 0)
+
+ // Returns empty array for empty inputs
+ override def nullable: Boolean = false
+
+ override def dataType: DataType = ArrayType(child.dataType, containsNull = child.nullable)
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(AnyDataType)
Review Comment:
seems we don't have any input type requirement, we don't need to extend `ImplicitCastInputTypes`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] zhengruifeng commented on a diff in pull request #38375: [SPARK-40900][SQL] Reimplement `frequentItems` with dataframe operations
Posted by GitBox <gi...@apache.org>.
zhengruifeng commented on code in PR #38375:
URL: https://github.com/apache/spark/pull/38375#discussion_r1004021114
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/stat/FrequentItems.scala:
##########
@@ -85,42 +51,144 @@ object FrequentItems extends Logging {
cols: Seq[String],
support: Double): DataFrame = {
require(support >= 1e-4 && support <= 1.0, s"Support must be in [1e-4, 1], but got $support.")
- val numCols = cols.length
+
// number of max items to keep counts for
val sizeOfMap = (1 / support).toInt
- val countMaps = Seq.tabulate(numCols)(i => new FreqItemCounter(sizeOfMap))
-
- val freqItems = df.select(cols.map(Column(_)) : _*).rdd.treeAggregate(countMaps)(
- seqOp = (counts, row) => {
- var i = 0
- while (i < numCols) {
- val thisMap = counts(i)
- val key = row.get(i)
- thisMap.add(key, 1L)
- i += 1
- }
- counts
- },
- combOp = (baseCounts, counts) => {
- var i = 0
- while (i < numCols) {
- baseCounts(i).merge(counts(i))
- i += 1
+
+ val frequentItemCols = cols.map { col =>
+ val aggExpr = new CollectFrequentItems(functions.col(col).expr, sizeOfMap)
+ Column(aggExpr.toAggregateExpression(isDistinct = false)).as(s"${col}_freqItems")
+ }
+
+ df.select(frequentItemCols: _*)
+ }
+}
+
+case class CollectFrequentItems(
+ child: Expression,
+ size: Int,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends TypedImperativeAggregate[mutable.Map[Any, Long]]
+ with ImplicitCastInputTypes with UnaryLike[Expression] {
+ require(size > 0)
+
+ def this(child: Expression, size: Int) = this(child, size, 0, 0)
+
+ // Returns empty array for empty inputs
+ override def nullable: Boolean = false
+
+ override def dataType: DataType = ArrayType(child.dataType, containsNull = child.nullable)
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(AnyDataType)
Review Comment:
good point, will update
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon closed pull request #38375: [SPARK-40900][SQL] Reimplement `frequentItems` with dataframe operations
Posted by GitBox <gi...@apache.org>.
HyukjinKwon closed pull request #38375: [SPARK-40900][SQL] Reimplement `frequentItems` with dataframe operations
URL: https://github.com/apache/spark/pull/38375
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] zhengruifeng commented on a diff in pull request #38375: [SPARK-40900][SQL] Reimplement `frequentItems` with dataframe operations
Posted by GitBox <gi...@apache.org>.
zhengruifeng commented on code in PR #38375:
URL: https://github.com/apache/spark/pull/38375#discussion_r1003976467
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/stat/FrequentItems.scala:
##########
@@ -85,42 +51,142 @@ object FrequentItems extends Logging {
cols: Seq[String],
support: Double): DataFrame = {
require(support >= 1e-4 && support <= 1.0, s"Support must be in [1e-4, 1], but got $support.")
- val numCols = cols.length
+
// number of max items to keep counts for
val sizeOfMap = (1 / support).toInt
- val countMaps = Seq.tabulate(numCols)(i => new FreqItemCounter(sizeOfMap))
-
- val freqItems = df.select(cols.map(Column(_)) : _*).rdd.treeAggregate(countMaps)(
- seqOp = (counts, row) => {
- var i = 0
- while (i < numCols) {
- val thisMap = counts(i)
- val key = row.get(i)
- thisMap.add(key, 1L)
- i += 1
- }
- counts
- },
- combOp = (baseCounts, counts) => {
- var i = 0
- while (i < numCols) {
- baseCounts(i).merge(counts(i))
- i += 1
+
+ val frequentItemCols = cols.map { col =>
+ val aggExpr = new CollectFrequentItems(functions.col(col).expr, sizeOfMap)
+ Column(aggExpr.toAggregateExpression(isDistinct = false)).as(s"${col}_freqItems")
+ }
+
+ df.select(frequentItemCols: _*)
+ }
+}
+
+case class CollectFrequentItems(
+ child: Expression,
+ size: Int,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends TypedImperativeAggregate[mutable.Map[Any, Long]]
+ with ImplicitCastInputTypes with UnaryLike[Expression] {
+ require(size > 0)
+
+ def this(child: Expression, size: Int) = this(child, size, 0, 0)
+
+ // Returns empty array for empty inputs
+ override def nullable: Boolean = false
+
+ override def dataType: DataType = ArrayType(child.dataType, containsNull = child.nullable)
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(AnyDataType)
+
+ override def prettyName: String = "collect_frequent_items"
+
+ override def createAggregationBuffer(): mutable.Map[Any, Long] =
+ mutable.Map.empty[Any, Long]
+
+ private def add(map: mutable.Map[Any, Long], key: Any, count: Long): mutable.Map[Any, Long] = {
+ if (map.contains(key)) {
+ map(key) += count
+ } else {
+ if (map.size < size) {
+ map += key -> count
+ } else {
+ val minCount = if (map.values.isEmpty) 0 else map.values.min
+ val remainder = count - minCount
+ if (remainder >= 0) {
+ map += key -> count // something will get kicked out, so we can add this
+ map.retain((k, v) => v > minCount)
+ map.transform((k, v) => v - minCount)
+ } else {
+ map.transform((k, v) => v - count)
}
- baseCounts
}
- )
- val justItems = freqItems.map(m => m.baseMap.keys.toArray)
- val resultRow = Row(justItems : _*)
+ }
+ map
+ }
+
+ override def update(
+ buffer: mutable.Map[Any, Long],
+ input: InternalRow): mutable.Map[Any, Long] = {
+ val key = child.eval(input)
+ if (key != null) {
+ this.add(buffer, InternalRow.copyValue(key), 1L)
+ } else {
+ this.add(buffer, key, 1L)
+ }
+ }
+
+ override def merge(
+ buffer: mutable.Map[Any, Long],
+ input: mutable.Map[Any, Long]): mutable.Map[Any, Long] = {
+ input.foreach { case (k, v) =>
+ add(buffer, k, v)
+ }
+ buffer
+ }
- val outputCols = cols.map { name =>
- val originalField = df.resolve(name)
+ override def eval(buffer: mutable.Map[Any, Long]): Any =
+ new GenericArrayData(buffer.keys.toArray)
- // append frequent Items to the column name for easy debugging
- StructField(name + "_freqItems", ArrayType(originalField.dataType, originalField.nullable))
- }.toArray
+ private lazy val projection =
+ UnsafeProjection.create(Array[DataType](child.dataType, LongType))
- val schema = StructType(outputCols).toAttributes
- Dataset.ofRows(df.sparkSession, LocalRelation.fromExternalRows(schema, Seq(resultRow)))
+ override def serialize(map: mutable.Map[Any, Long]): Array[Byte] = {
+ val buffer = new Array[Byte](4 << 10) // 4K
+ val bos = new ByteArrayOutputStream()
+ val out = new DataOutputStream(bos)
+ try {
Review Comment:
good point, will update
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] zhengruifeng commented on a diff in pull request #38375: [SPARK-40900][SQL] Reimplement `frequentItems` with dataframe operations
Posted by GitBox <gi...@apache.org>.
zhengruifeng commented on code in PR #38375:
URL: https://github.com/apache/spark/pull/38375#discussion_r1003976380
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/stat/FrequentItems.scala:
##########
@@ -85,42 +51,142 @@ object FrequentItems extends Logging {
cols: Seq[String],
support: Double): DataFrame = {
require(support >= 1e-4 && support <= 1.0, s"Support must be in [1e-4, 1], but got $support.")
- val numCols = cols.length
+
// number of max items to keep counts for
val sizeOfMap = (1 / support).toInt
- val countMaps = Seq.tabulate(numCols)(i => new FreqItemCounter(sizeOfMap))
-
- val freqItems = df.select(cols.map(Column(_)) : _*).rdd.treeAggregate(countMaps)(
- seqOp = (counts, row) => {
- var i = 0
- while (i < numCols) {
- val thisMap = counts(i)
- val key = row.get(i)
- thisMap.add(key, 1L)
- i += 1
- }
- counts
- },
- combOp = (baseCounts, counts) => {
- var i = 0
- while (i < numCols) {
- baseCounts(i).merge(counts(i))
- i += 1
+
+ val frequentItemCols = cols.map { col =>
+ val aggExpr = new CollectFrequentItems(functions.col(col).expr, sizeOfMap)
+ Column(aggExpr.toAggregateExpression(isDistinct = false)).as(s"${col}_freqItems")
+ }
+
+ df.select(frequentItemCols: _*)
+ }
+}
+
+case class CollectFrequentItems(
+ child: Expression,
+ size: Int,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends TypedImperativeAggregate[mutable.Map[Any, Long]]
+ with ImplicitCastInputTypes with UnaryLike[Expression] {
+ require(size > 0)
+
+ def this(child: Expression, size: Int) = this(child, size, 0, 0)
+
+ // Returns empty array for empty inputs
+ override def nullable: Boolean = false
+
+ override def dataType: DataType = ArrayType(child.dataType, containsNull = child.nullable)
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(AnyDataType)
+
+ override def prettyName: String = "collect_frequent_items"
+
+ override def createAggregationBuffer(): mutable.Map[Any, Long] =
+ mutable.Map.empty[Any, Long]
+
+ private def add(map: mutable.Map[Any, Long], key: Any, count: Long): mutable.Map[Any, Long] = {
+ if (map.contains(key)) {
+ map(key) += count
+ } else {
+ if (map.size < size) {
+ map += key -> count
+ } else {
+ val minCount = if (map.values.isEmpty) 0 else map.values.min
+ val remainder = count - minCount
+ if (remainder >= 0) {
+ map += key -> count // something will get kicked out, so we can add this
+ map.retain((k, v) => v > minCount)
+ map.transform((k, v) => v - minCount)
+ } else {
+ map.transform((k, v) => v - count)
}
- baseCounts
}
- )
- val justItems = freqItems.map(m => m.baseMap.keys.toArray)
- val resultRow = Row(justItems : _*)
+ }
+ map
+ }
+
+ override def update(
+ buffer: mutable.Map[Any, Long],
+ input: InternalRow): mutable.Map[Any, Long] = {
+ val key = child.eval(input)
+ if (key != null) {
+ this.add(buffer, InternalRow.copyValue(key), 1L)
+ } else {
+ this.add(buffer, key, 1L)
+ }
+ }
+
+ override def merge(
+ buffer: mutable.Map[Any, Long],
+ input: mutable.Map[Any, Long]): mutable.Map[Any, Long] = {
+ input.foreach { case (k, v) =>
+ add(buffer, k, v)
+ }
+ buffer
+ }
- val outputCols = cols.map { name =>
- val originalField = df.resolve(name)
+ override def eval(buffer: mutable.Map[Any, Long]): Any =
+ new GenericArrayData(buffer.keys.toArray)
Review Comment:
I guess it is only invoked once, so no need to reuse?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] zhengruifeng commented on a diff in pull request #38375: [SPARK-40900][SQL] Reimplement `frequentItems` with dataframe operations
Posted by GitBox <gi...@apache.org>.
zhengruifeng commented on code in PR #38375:
URL: https://github.com/apache/spark/pull/38375#discussion_r1003991925
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/stat/FrequentItems.scala:
##########
@@ -85,42 +51,142 @@ object FrequentItems extends Logging {
cols: Seq[String],
support: Double): DataFrame = {
require(support >= 1e-4 && support <= 1.0, s"Support must be in [1e-4, 1], but got $support.")
- val numCols = cols.length
+
// number of max items to keep counts for
val sizeOfMap = (1 / support).toInt
- val countMaps = Seq.tabulate(numCols)(i => new FreqItemCounter(sizeOfMap))
-
- val freqItems = df.select(cols.map(Column(_)) : _*).rdd.treeAggregate(countMaps)(
- seqOp = (counts, row) => {
- var i = 0
- while (i < numCols) {
- val thisMap = counts(i)
- val key = row.get(i)
- thisMap.add(key, 1L)
- i += 1
- }
- counts
- },
- combOp = (baseCounts, counts) => {
- var i = 0
- while (i < numCols) {
- baseCounts(i).merge(counts(i))
- i += 1
+
+ val frequentItemCols = cols.map { col =>
+ val aggExpr = new CollectFrequentItems(functions.col(col).expr, sizeOfMap)
+ Column(aggExpr.toAggregateExpression(isDistinct = false)).as(s"${col}_freqItems")
+ }
+
+ df.select(frequentItemCols: _*)
+ }
+}
+
+case class CollectFrequentItems(
+ child: Expression,
+ size: Int,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends TypedImperativeAggregate[mutable.Map[Any, Long]]
+ with ImplicitCastInputTypes with UnaryLike[Expression] {
+ require(size > 0)
+
+ def this(child: Expression, size: Int) = this(child, size, 0, 0)
+
+ // Returns empty array for empty inputs
+ override def nullable: Boolean = false
+
+ override def dataType: DataType = ArrayType(child.dataType, containsNull = child.nullable)
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(AnyDataType)
+
+ override def prettyName: String = "collect_frequent_items"
+
+ override def createAggregationBuffer(): mutable.Map[Any, Long] =
+ mutable.Map.empty[Any, Long]
+
+ private def add(map: mutable.Map[Any, Long], key: Any, count: Long): mutable.Map[Any, Long] = {
+ if (map.contains(key)) {
+ map(key) += count
+ } else {
+ if (map.size < size) {
+ map += key -> count
+ } else {
+ val minCount = if (map.values.isEmpty) 0 else map.values.min
+ val remainder = count - minCount
+ if (remainder >= 0) {
+ map += key -> count // something will get kicked out, so we can add this
+ map.retain((k, v) => v > minCount)
+ map.transform((k, v) => v - minCount)
+ } else {
+ map.transform((k, v) => v - count)
}
- baseCounts
}
- )
- val justItems = freqItems.map(m => m.baseMap.keys.toArray)
- val resultRow = Row(justItems : _*)
+ }
+ map
+ }
+
+ override def update(
+ buffer: mutable.Map[Any, Long],
+ input: InternalRow): mutable.Map[Any, Long] = {
+ val key = child.eval(input)
+ if (key != null) {
+ this.add(buffer, InternalRow.copyValue(key), 1L)
+ } else {
+ this.add(buffer, key, 1L)
+ }
+ }
+
+ override def merge(
+ buffer: mutable.Map[Any, Long],
+ input: mutable.Map[Any, Long]): mutable.Map[Any, Long] = {
+ input.foreach { case (k, v) =>
+ add(buffer, k, v)
+ }
+ buffer
+ }
- val outputCols = cols.map { name =>
- val originalField = df.resolve(name)
+ override def eval(buffer: mutable.Map[Any, Long]): Any =
+ new GenericArrayData(buffer.keys.toArray)
- // append frequent Items to the column name for easy debugging
- StructField(name + "_freqItems", ArrayType(originalField.dataType, originalField.nullable))
- }.toArray
+ private lazy val projection =
+ UnsafeProjection.create(Array[DataType](child.dataType, LongType))
- val schema = StructType(outputCols).toAttributes
- Dataset.ofRows(df.sparkSession, LocalRelation.fromExternalRows(schema, Seq(resultRow)))
+ override def serialize(map: mutable.Map[Any, Long]): Array[Byte] = {
+ val buffer = new Array[Byte](4 << 10) // 4K
+ val bos = new ByteArrayOutputStream()
+ val out = new DataOutputStream(bos)
+ try {
Review Comment:
seem we can not leverage `Utils.tryWithResource` here, since `Utils.tryWithResource` only support single `Closeable` but there are two ones `bos` and `out`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on pull request #38375: [SPARK-40900][SQL] Reimplement `frequentItems` with dataframe operations
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on PR #38375:
URL: https://github.com/apache/spark/pull/38375#issuecomment-1291289206
Merged to master.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org