Github user guowei2 commented on the pull request:
https://github.com/apache/spark/pull/2029#issuecomment-53426380
```
import org.apache.spark.sql.catalyst.types.{IntegerType, DataType}
import org.apache.spark.sql.catalyst.dsl.expressions._
import org.apache.spark.sql.execution._
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.SparkContext._
import org.apache.spark._
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.execution.OnHeapAggregate
import org.apache.spark.sql.catalyst.expressions.Alias
import org.apache.spark.sql.catalyst.expressions.BoundReference
object AggregateBenchMark extends App {
val sc = new SparkContext(
new SparkConf().setMaster("local").setAppName("agg-benchmark"))
val dataType: DataType = IntegerType
val aggExps = Seq(Alias(sum(BoundReference(1, dataType, true)),"sum")())
val groupExps = Seq(BoundReference(0, dataType, true))
val attributes = aggExps.map(_.toAttribute)
val childPlan = rowsPlan(sc, attributes)
def benchmarkOnHeap = {
val begin = System.currentTimeMillis()
OnHeapAggregate(false, groupExps, aggExps,
childPlan).execute().foreach(_ => {})
val end = System.currentTimeMillis()
end - begin
}
def benchmarkExternal = {
val begin = System.currentTimeMillis()
ExternalAggregate(false, groupExps, aggExps,
childPlan).execute().foreach(_ => {})
val end = System.currentTimeMillis()
end - begin
}
(1 to 5).map(_=> println("OnHeapAggregate time: "+ benchmarkOnHeap))
(1 to 5).map(_=> println("ExternalAggregate time: "+ benchmarkExternal))
}
private[spark] class TestRDD(
sc: SparkContext,
numPartitions: Int) extends RDD[Row](sc, Nil) with Serializable {
override def compute(split: Partition, context: TaskContext):
Iterator[Row] = {
new Iterator[Row] {
var lines = 0
override final def hasNext: Boolean = lines < 3000000
override final def next(): Row = {
lines += 1
val row = new GenericMutableRow(2)
row(0) = (math.random * 2000).toInt
row(1) = (math.random * 50).toInt
row.asInstanceOf[Row]
}
}
}
override def getPartitions = (0 until numPartitions).map(i => new
Partition {
override def index = i
}).toArray
override def getPreferredLocations(split: Partition): Seq[String] = Nil
override def toString: String = "TestRDD " + id
}
case class rowsPlan(@transient val sc:SparkContext, attributes:
Seq[Attribute]) extends LeafNode {
override def output = attributes
override def execute() = {
new TestRDD(sc, 1).asInstanceOf[RDD[Row]]
}
}
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
