You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2016/07/12 09:17:59 UTC
spark git commit: [SPARK-16189][SQL] Add ExternalRDD logical plan for
input with RDD to have a chance to eliminate serialize/deserialize.
Repository: spark
Updated Branches:
refs/heads/master fc11c509e -> 5b28e0258
[SPARK-16189][SQL] Add ExternalRDD logical plan for input with RDD to have a chance to eliminate serialize/deserialize.
## What changes were proposed in this pull request?
Currently the input `RDD` of `Dataset` is always serialized to `RDD[InternalRow]` prior to being as `Dataset`, but there is a case that we use `map` or `mapPartitions` just after converted to `Dataset`.
In this case, serialize and then deserialize happens but it would not be needed.
This pr adds `ExistingRDD` logical plan for input with `RDD` to have a chance to eliminate serialize/deserialize.
## How was this patch tested?
Existing tests.
Author: Takuya UESHIN <ue...@happy-camper.st>
Closes #13890 from ueshin/issues/SPARK-16189.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/5b28e025
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/5b28e025
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/5b28e025
Branch: refs/heads/master
Commit: 5b28e02584fa4da85214e7da6d77b3b8e189b781
Parents: fc11c50
Author: Takuya UESHIN <ue...@happy-camper.st>
Authored: Tue Jul 12 17:16:59 2016 +0800
Committer: Wenchen Fan <we...@databricks.com>
Committed: Tue Jul 12 17:16:59 2016 +0800
----------------------------------------------------------------------
.../org/apache/spark/sql/SparkSession.scala | 12 +---
.../spark/sql/execution/ExistingRDD.scala | 72 +++++++++++++++++---
.../sql/execution/LocalTableScanExec.scala | 8 ++-
.../spark/sql/execution/SparkStrategies.scala | 1 +
.../scala/org/apache/spark/sql/QueryTest.scala | 8 ++-
5 files changed, 81 insertions(+), 20 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/5b28e025/sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala
----------------------------------------------------------------------
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala b/sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala
index a3fd39d..1271d1c 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala
@@ -237,10 +237,8 @@ class SparkSession private(
@Experimental
def createDataFrame[A <: Product : TypeTag](rdd: RDD[A]): DataFrame = {
SparkSession.setActiveSession(this)
- val schema = ScalaReflection.schemaFor[A].dataType.asInstanceOf[StructType]
- val attributeSeq = schema.toAttributes
- val rowRDD = RDDConversions.productToRowRdd(rdd, schema.map(_.dataType))
- Dataset.ofRows(self, LogicalRDD(attributeSeq, rowRDD)(self))
+ val encoder = Encoders.product[A]
+ Dataset.ofRows(self, ExternalRDD(rdd, self)(encoder))
}
/**
@@ -425,11 +423,7 @@ class SparkSession private(
*/
@Experimental
def createDataset[T : Encoder](data: RDD[T]): Dataset[T] = {
- val enc = encoderFor[T]
- val attributes = enc.schema.toAttributes
- val encoded = data.map(d => enc.toRow(d))
- val plan = LogicalRDD(attributes, encoded)(self)
- Dataset[T](self, plan)
+ Dataset[T](self, ExternalRDD(data, self))
}
/**
http://git-wip-us.apache.org/repos/asf/spark/blob/5b28e025/sql/core/src/main/scala/org/apache/spark/sql/execution/ExistingRDD.scala
----------------------------------------------------------------------
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/ExistingRDD.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/ExistingRDD.scala
index 09203e6..491c274 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/ExistingRDD.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/ExistingRDD.scala
@@ -20,12 +20,12 @@ package org.apache.spark.sql.execution
import org.apache.commons.lang3.StringUtils
import org.apache.spark.rdd.RDD
-import org.apache.spark.sql.{AnalysisException, Row, SparkSession, SQLContext}
+import org.apache.spark.sql.{AnalysisException, Encoder, Row, SparkSession, SQLContext}
import org.apache.spark.sql.catalyst.{CatalystTypeConverters, InternalRow, TableIdentifier}
import org.apache.spark.sql.catalyst.analysis.MultiInstanceRelation
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext, ExprCode}
-import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Statistics}
+import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.plans.physical.{HashPartitioning, Partitioning, UnknownPartitioning}
import org.apache.spark.sql.execution.datasources.HadoopFsRelation
import org.apache.spark.sql.execution.datasources.parquet.{ParquetFileFormat => ParquetSource}
@@ -74,13 +74,71 @@ object RDDConversions {
}
}
+private[sql] object ExternalRDD {
+
+ def apply[T: Encoder](rdd: RDD[T], session: SparkSession): LogicalPlan = {
+ val externalRdd = ExternalRDD(CatalystSerde.generateObjAttr[T], rdd)(session)
+ CatalystSerde.serialize[T](externalRdd)
+ }
+}
+
/** Logical plan node for scanning data from an RDD. */
+private[sql] case class ExternalRDD[T](
+ outputObjAttr: Attribute,
+ rdd: RDD[T])(session: SparkSession)
+ extends LeafNode with ObjectProducer with MultiInstanceRelation {
+
+ override protected final def otherCopyArgs: Seq[AnyRef] = session :: Nil
+
+ override def newInstance(): ExternalRDD.this.type =
+ ExternalRDD(outputObjAttr.newInstance(), rdd)(session).asInstanceOf[this.type]
+
+ override def sameResult(plan: LogicalPlan): Boolean = {
+ plan.canonicalized match {
+ case ExternalRDD(_, otherRDD) => rdd.id == otherRDD.id
+ case _ => false
+ }
+ }
+
+ override protected def stringArgs: Iterator[Any] = Iterator(output)
+
+ @transient override lazy val statistics: Statistics = Statistics(
+ // TODO: Instead of returning a default value here, find a way to return a meaningful size
+ // estimate for RDDs. See PR 1238 for more discussions.
+ sizeInBytes = BigInt(session.sessionState.conf.defaultSizeInBytes)
+ )
+}
+
+/** Physical plan node for scanning data from an RDD. */
+private[sql] case class ExternalRDDScanExec[T](
+ outputObjAttr: Attribute,
+ rdd: RDD[T]) extends LeafExecNode with ObjectProducerExec {
+
+ private[sql] override lazy val metrics = Map(
+ "numOutputRows" -> SQLMetrics.createMetric(sparkContext, "number of output rows"))
+
+ protected override def doExecute(): RDD[InternalRow] = {
+ val numOutputRows = longMetric("numOutputRows")
+ val outputDataType = outputObjAttr.dataType
+ rdd.mapPartitionsInternal { iter =>
+ val outputObject = ObjectOperator.wrapObjectToRow(outputDataType)
+ iter.map { value =>
+ numOutputRows += 1
+ outputObject(value)
+ }
+ }
+ }
+
+ override def simpleString: String = {
+ s"Scan $nodeName${output.mkString("[", ",", "]")}"
+ }
+}
+
+/** Logical plan node for scanning data from an RDD of InternalRow. */
private[sql] case class LogicalRDD(
output: Seq[Attribute],
rdd: RDD[InternalRow])(session: SparkSession)
- extends LogicalPlan with MultiInstanceRelation {
-
- override def children: Seq[LogicalPlan] = Nil
+ extends LeafNode with MultiInstanceRelation {
override protected final def otherCopyArgs: Seq[AnyRef] = session :: Nil
@@ -96,8 +154,6 @@ private[sql] case class LogicalRDD(
override protected def stringArgs: Iterator[Any] = Iterator(output)
- override def producedAttributes: AttributeSet = outputSet
-
@transient override lazy val statistics: Statistics = Statistics(
// TODO: Instead of returning a default value here, find a way to return a meaningful size
// estimate for RDDs. See PR 1238 for more discussions.
@@ -105,7 +161,7 @@ private[sql] case class LogicalRDD(
)
}
-/** Physical plan node for scanning data from an RDD. */
+/** Physical plan node for scanning data from an RDD of InternalRow. */
private[sql] case class RDDScanExec(
output: Seq[Attribute],
rdd: RDD[InternalRow],
http://git-wip-us.apache.org/repos/asf/spark/blob/5b28e025/sql/core/src/main/scala/org/apache/spark/sql/execution/LocalTableScanExec.scala
----------------------------------------------------------------------
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/LocalTableScanExec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/LocalTableScanExec.scala
index df2f238..f86f42b 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/LocalTableScanExec.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/LocalTableScanExec.scala
@@ -34,8 +34,12 @@ private[sql] case class LocalTableScanExec(
"numOutputRows" -> SQLMetrics.createMetric(sparkContext, "number of output rows"))
private val unsafeRows: Array[InternalRow] = {
- val proj = UnsafeProjection.create(output, output)
- rows.map(r => proj(r).copy()).toArray
+ if (rows.isEmpty) {
+ Array.empty
+ } else {
+ val proj = UnsafeProjection.create(output, output)
+ rows.map(r => proj(r).copy()).toArray
+ }
}
private lazy val rdd = sqlContext.sparkContext.parallelize(unsafeRows)
http://git-wip-us.apache.org/repos/asf/spark/blob/5b28e025/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
----------------------------------------------------------------------
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
index 5e643ea..52e1981 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
@@ -411,6 +411,7 @@ private[sql] abstract class SparkStrategies extends QueryPlanner[SparkPlan] {
case logical.RepartitionByExpression(expressions, child, nPartitions) =>
exchange.ShuffleExchange(HashPartitioning(
expressions, nPartitions.getOrElse(numPartitions)), planLater(child)) :: Nil
+ case ExternalRDD(outputObjAttr, rdd) => ExternalRDDScanExec(outputObjAttr, rdd) :: Nil
case LogicalRDD(output, rdd) => RDDScanExec(output, rdd, "ExistingRDD") :: Nil
case BroadcastHint(child) => planLater(child) :: Nil
case _ => Nil
http://git-wip-us.apache.org/repos/asf/spark/blob/5b28e025/sql/core/src/test/scala/org/apache/spark/sql/QueryTest.scala
----------------------------------------------------------------------
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/QueryTest.scala b/sql/core/src/test/scala/org/apache/spark/sql/QueryTest.scala
index ab50513..a9d0fcf 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/QueryTest.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/QueryTest.scala
@@ -28,7 +28,7 @@ import org.apache.spark.sql.catalyst.plans._
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.trees.TreeNode
import org.apache.spark.sql.catalyst.util._
-import org.apache.spark.sql.execution.LogicalRDD
+import org.apache.spark.sql.execution._
import org.apache.spark.sql.execution.aggregate.TypedAggregateExpression
import org.apache.spark.sql.execution.columnar.InMemoryRelation
import org.apache.spark.sql.execution.datasources.LogicalRelation
@@ -242,6 +242,12 @@ abstract class QueryTest extends PlanTest {
case _: LogicalRelation => return
case p if p.getClass.getSimpleName == "MetastoreRelation" => return
case _: MemoryPlan => return
+ case p: InMemoryRelation =>
+ p.child.transform {
+ case _: ObjectConsumerExec => return
+ case _: ObjectProducerExec => return
+ }
+ p
}.transformAllExpressions {
case a: ImperativeAggregate => return
case _: TypedAggregateExpression => return
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org