Upgrade to latest spark versions 3.3.0, 3.2.2 and 3.1.3 (#415)

Emil Ejbyfeldt · web-flow · commit 100fb01fb5c6 · 2022-08-25T08:32:59.000+08:00
* Upgrade to latest spark versions

Add CI build for spark 3.3.0

Upgrade spark 3.2.0 to 3.2.2

Upgrade spark 3.1.2 to 3.1.3

* Share single SparkContext for all tests

The previous setup would cause test to end up using a stopped spark
context causing them to fail. This seems to be due to some change in
spark 3.3.0, by making the whole suite using a single SparkContext this
issue goes away.
diff --git a/.github/workflows/python-ci.yml b/.github/workflows/python-ci.yml
@@ -6,9 +6,11 @@ jobs:
       fail-fast: false
       matrix:
         include:
-          - spark-version: 3.2.0
+          - spark-version: 3.3.0
             scala-version: 2.12.12
-          - spark-version: 3.1.2
+          - spark-version: 3.2.2
+            scala-version: 2.12.12
+          - spark-version: 3.1.3
             scala-version: 2.12.12
             python-version: 3.8
           - spark-version: 3.0.3
diff --git a/.github/workflows/scala-ci.yml b/.github/workflows/scala-ci.yml
@@ -6,9 +6,11 @@ jobs:
       fail-fast: false
       matrix:
         include:
-          - spark-version: 3.2.0
+          - spark-version: 3.3.0
             scala-version: 2.12.12
-          - spark-version: 3.1.2
+          - spark-version: 3.2.2
+            scala-version: 2.12.12
+          - spark-version: 3.1.3
             scala-version: 2.12.12
           - spark-version: 3.0.3
             scala-version: 2.12.12
diff --git a/Dockerfile b/Dockerfile
@@ -9,7 +9,7 @@ RUN apt-get update && \
 
 # Install Spark and update env variables.
 ENV SCALA_VERSION 2.12.15
-ENV SPARK_VERSION "3.2.0"
+ENV SPARK_VERSION "3.3.0"
 ENV SPARK_BUILD "spark-${SPARK_VERSION}-bin-hadoop3.2"
 ENV SPARK_BUILD_URL "https://dist.apache.org/repos/dist/release/spark/spark-${SPARK_VERSION}/${SPARK_BUILD}.tgz"
 RUN wget --quiet "$SPARK_BUILD_URL" -O /tmp/spark.tgz && \
diff --git a/build.sbt b/build.sbt
@@ -5,9 +5,10 @@ import ReleaseTransformations._
 
 resolvers += "Spark snapshot repository" at "https://repository.apache.org/snapshots/"
 
-val sparkVer = sys.props.getOrElse("spark.version", "3.2.0")
+val sparkVer = sys.props.getOrElse("spark.version", "3.3.0")
 val sparkBranch = sparkVer.substring(0, 3)
 val defaultScalaVer = sparkBranch match {
+  case "3.3" => "2.12.15"
   case "3.2" => "2.12.15"
   case "3.1" => "2.12.15"
   case "3.0" => "2.12.15"
diff --git a/dev/release.py b/dev/release.py
@@ -39,7 +39,7 @@ def verify(prompt, interactive):
 @click.option("--publish-docs", type=bool, default=PUBLISH_DOCS_DEFAULT, show_default=True,
               help="Publish docs to github-pages.")
 @click.option("--spark-version", multiple=True, show_default=True,
-              default=["2.4.8", "3.0.3", "3.1.2", "3.2.0"])
+              default=["2.4.8", "3.0.3", "3.1.3", "3.2.2", "3.3.0"])
 def main(release_version, next_version, publish_to, no_prompt, git_remote, publish_docs,
          spark_version):
     interactive = not no_prompt
diff --git a/python/graphframes/tests.py b/python/graphframes/tests.py
@@ -67,10 +67,18 @@ def parse_spark_version(cls, version_str):
         return version_info
 
     @classmethod
-    def register(cls, sc):
-        cls.sc = sc
+    def createSparkContext(cls):
+        cls.sc = sc = SparkContext('local[4]', "GraphFramesTests")
+        cls.checkpointDir = tempfile.mkdtemp()
+        cls.sc.setCheckpointDir(cls.checkpointDir)
         cls.spark_version = cls.parse_spark_version(sc.version)
 
+    @classmethod
+    def stopSparkContext(cls):
+        cls.sc.stop()
+        cls.sc = None
+        shutil.rmtree(cls.checkpointDir)
+
     @classmethod
     def spark_at_least_of_version(cls, version_str):
         assert hasattr(cls, 'spark_version')
@@ -84,25 +92,24 @@ def spark_at_least_of_version(cls, version_str):
         # All major.minor.maintenance equal
         return True
 
+def setUpModule():
+    GraphFrameTestUtils.createSparkContext()
+
+def tearDownModule():
+    GraphFrameTestUtils.stopSparkContext()
+
 
 class GraphFrameTestCase(unittest.TestCase):
 
     @classmethod
     def setUpClass(cls):
-        cls.sc = SparkContext('local[4]', cls.__name__)
-        cls.checkpointDir = tempfile.mkdtemp()
-        cls.sc.setCheckpointDir(cls.checkpointDir)
-        cls.sql = SQLContext(cls.sc)
+        cls.sql = SQLContext(GraphFrameTestUtils.sc)
         # Small tests run much faster with spark.sql.shuffle.partitions=4
         cls.sql.setConf("spark.sql.shuffle.partitions", "4")
-        GraphFrameTestUtils.register(cls.sc)
 
     @classmethod
     def tearDownClass(cls):
-        cls.sc.stop()
-        cls.sc = None
         cls.sql = None
-        shutil.rmtree(cls.checkpointDir)
 
 
 class GraphFrameTest(GraphFrameTestCase):
