spark-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sro...@apache.org
Subject spark git commit: [SPARK-21977][HOTFIX] Adjust EnsureStatefulOpPartitioningSuite to use scalatest lifecycle normally instead of constructor
Date Thu, 21 Sep 2017 17:00:22 GMT
Repository: spark
Updated Branches:
  refs/heads/master 1270e7175 -> f10cbf17d


[SPARK-21977][HOTFIX] Adjust EnsureStatefulOpPartitioningSuite to use scalatest lifecycle
normally instead of constructor

## What changes were proposed in this pull request?

Adjust EnsureStatefulOpPartitioningSuite to use scalatest lifecycle normally instead of constructor;
fixes:

```
*** RUN ABORTED ***
  org.apache.spark.SparkException: Only one SparkContext may be running in this JVM (see SPARK-2243).
To ignore this error, set spark.driver.allowMultipleContexts = true. The currently running
SparkContext was created at:
org.apache.spark.sql.streaming.EnsureStatefulOpPartitioningSuite.<init>(EnsureStatefulOpPartitioningSuite.scala:35)
```

## How was this patch tested?

Existing tests

Author: Sean Owen <sowen@cloudera.com>

Closes #19306 from srowen/SPARK-21977.2.


Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/f10cbf17
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/f10cbf17
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/f10cbf17

Branch: refs/heads/master
Commit: f10cbf17dc7ceb96982fcdc964c849336fb50deb
Parents: 1270e71
Author: Sean Owen <sowen@cloudera.com>
Authored: Thu Sep 21 18:00:19 2017 +0100
Committer: Sean Owen <sowen@cloudera.com>
Committed: Thu Sep 21 18:00:19 2017 +0100

----------------------------------------------------------------------
 .../EnsureStatefulOpPartitioningSuite.scala     | 92 +++++++++++---------
 1 file changed, 49 insertions(+), 43 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/spark/blob/f10cbf17/sql/core/src/test/scala/org/apache/spark/sql/streaming/EnsureStatefulOpPartitioningSuite.scala
----------------------------------------------------------------------
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/EnsureStatefulOpPartitioningSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/streaming/EnsureStatefulOpPartitioningSuite.scala
index 66c0263..044bb03 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/EnsureStatefulOpPartitioningSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/EnsureStatefulOpPartitioningSuite.scala
@@ -20,6 +20,7 @@ package org.apache.spark.sql.streaming
 import java.util.UUID
 
 import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.DataFrame
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.analysis.UnresolvedAttribute
 import org.apache.spark.sql.catalyst.expressions.Attribute
@@ -32,39 +33,47 @@ import org.apache.spark.sql.test.SharedSQLContext
 class EnsureStatefulOpPartitioningSuite extends SparkPlanTest with SharedSQLContext {
 
   import testImplicits._
-  super.beforeAll()
 
-  private val baseDf = Seq((1, "A"), (2, "b")).toDF("num", "char")
+  private var baseDf: DataFrame = null
 
-  testEnsureStatefulOpPartitioning(
-    "ClusteredDistribution generates Exchange with HashPartitioning",
-    baseDf.queryExecution.sparkPlan,
-    requiredDistribution = keys => ClusteredDistribution(keys),
-    expectedPartitioning =
-      keys => HashPartitioning(keys, spark.sessionState.conf.numShufflePartitions),
-    expectShuffle = true)
+  override def beforeAll(): Unit = {
+    super.beforeAll()
+    baseDf = Seq((1, "A"), (2, "b")).toDF("num", "char")
+  }
+
+  test("ClusteredDistribution generates Exchange with HashPartitioning") {
+    testEnsureStatefulOpPartitioning(
+      baseDf.queryExecution.sparkPlan,
+      requiredDistribution = keys => ClusteredDistribution(keys),
+      expectedPartitioning =
+        keys => HashPartitioning(keys, spark.sessionState.conf.numShufflePartitions),
+      expectShuffle = true)
+  }
 
-  testEnsureStatefulOpPartitioning(
-    "ClusteredDistribution with coalesce(1) generates Exchange with HashPartitioning",
-    baseDf.coalesce(1).queryExecution.sparkPlan,
-    requiredDistribution = keys => ClusteredDistribution(keys),
-    expectedPartitioning =
-      keys => HashPartitioning(keys, spark.sessionState.conf.numShufflePartitions),
-    expectShuffle = true)
+  test("ClusteredDistribution with coalesce(1) generates Exchange with HashPartitioning")
{
+    testEnsureStatefulOpPartitioning(
+      baseDf.coalesce(1).queryExecution.sparkPlan,
+      requiredDistribution = keys => ClusteredDistribution(keys),
+      expectedPartitioning =
+        keys => HashPartitioning(keys, spark.sessionState.conf.numShufflePartitions),
+      expectShuffle = true)
+  }
 
-  testEnsureStatefulOpPartitioning(
-    "AllTuples generates Exchange with SinglePartition",
-    baseDf.queryExecution.sparkPlan,
-    requiredDistribution = _ => AllTuples,
-    expectedPartitioning = _ => SinglePartition,
-    expectShuffle = true)
+  test("AllTuples generates Exchange with SinglePartition") {
+    testEnsureStatefulOpPartitioning(
+      baseDf.queryExecution.sparkPlan,
+      requiredDistribution = _ => AllTuples,
+      expectedPartitioning = _ => SinglePartition,
+      expectShuffle = true)
+  }
 
-  testEnsureStatefulOpPartitioning(
-    "AllTuples with coalesce(1) doesn't need Exchange",
-    baseDf.coalesce(1).queryExecution.sparkPlan,
-    requiredDistribution = _ => AllTuples,
-    expectedPartitioning = _ => SinglePartition,
-    expectShuffle = false)
+  test("AllTuples with coalesce(1) doesn't need Exchange") {
+    testEnsureStatefulOpPartitioning(
+      baseDf.coalesce(1).queryExecution.sparkPlan,
+      requiredDistribution = _ => AllTuples,
+      expectedPartitioning = _ => SinglePartition,
+      expectShuffle = false)
+  }
 
   /**
    * For `StatefulOperator` with the given `requiredChildDistribution`, and child SparkPlan
@@ -72,26 +81,23 @@ class EnsureStatefulOpPartitioningSuite extends SparkPlanTest with SharedSQLCont
    * ensure the expected partitioning.
    */
   private def testEnsureStatefulOpPartitioning(
-      testName: String,
       inputPlan: SparkPlan,
       requiredDistribution: Seq[Attribute] => Distribution,
       expectedPartitioning: Seq[Attribute] => Partitioning,
       expectShuffle: Boolean): Unit = {
-    test(testName) {
-      val operator = TestStatefulOperator(inputPlan, requiredDistribution(inputPlan.output.take(1)))
-      val executed = executePlan(operator, OutputMode.Complete())
-      if (expectShuffle) {
-        val exchange = executed.children.find(_.isInstanceOf[Exchange])
-        if (exchange.isEmpty) {
-          fail(s"Was expecting an exchange but didn't get one in:\n$executed")
-        }
-        assert(exchange.get ===
-          ShuffleExchange(expectedPartitioning(inputPlan.output.take(1)), inputPlan),
-          s"Exchange didn't have expected properties:\n${exchange.get}")
-      } else {
-        assert(!executed.children.exists(_.isInstanceOf[Exchange]),
-          s"Unexpected exchange found in:\n$executed")
+    val operator = TestStatefulOperator(inputPlan, requiredDistribution(inputPlan.output.take(1)))
+    val executed = executePlan(operator, OutputMode.Complete())
+    if (expectShuffle) {
+      val exchange = executed.children.find(_.isInstanceOf[Exchange])
+      if (exchange.isEmpty) {
+        fail(s"Was expecting an exchange but didn't get one in:\n$executed")
       }
+      assert(exchange.get ===
+        ShuffleExchange(expectedPartitioning(inputPlan.output.take(1)), inputPlan),
+        s"Exchange didn't have expected properties:\n${exchange.get}")
+    } else {
+      assert(!executed.children.exists(_.isInstanceOf[Exchange]),
+        s"Unexpected exchange found in:\n$executed")
     }
   }
 


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org


Mime
View raw message