title: How do I create a shared SparkSession fixture in scalatest that persists between test suites?
body: <p>I am very new to Scala and Scalatest, but have some experience with Pyspark, and I'm trying to learn Scala from a Spark perspective.</p>
  <p>I'm currently trying to get my head around exactly the correct way to set up and use fixtures within Scalatest.</p>
  <p>The way I imagine this working, and this may not be how this is done in Scala, would be that I would setup a <code>SparkSession</code> as a global fixture shared amongst suites of test, then with potentially several sample datasets hooking into that <code>SparkSession</code> that may be used for an individual test suite with several tests etc.</p>
  <p>At the moment, I have some code that is working for running several tests in the same suite using the shared fixture using the <code>BeforeAndAfterAll</code> trait; however, if I run several suites at the same time, the suite that completes first appears to terminate the <code>SparkSession</code> and any further tests fail with <code>java.lang.IllegalStateException: Cannot call methods on a stopped SparkContext.</code></p>
  <p>So, I was wondering if there is a way to create the <code>SparkSession</code> so that it will only be stopped when <em>all</em> running suites have completed; or if I'm barking up the wrong tree and there is a better approach altogether - as I say, I'm very new to Scala so this might just not be how you do this, in which case alternative suggestions are very welcome.</p>
  <p>First I have a package testSetup and I'm creating a trait for the SparkSession:</p>
  <pre class="lang-scala prettyprint-override"><code>package com.example.test
  
  import org.apache.spark.sql.SparkSession
  import org.apache.spark.sql.DataFrame
  import org.apache.spark.sql.Row
  import org.apache.spark.sql.types._
  
  import org.scalatest._
  import org.scalatest.FixtureSuite
  import org.scalatest.funsuite.FixtureAnyFunSuite
  
  package testSetup {
  
   trait SparkSetup  {
      val spark = SparkSession
        .builder
        .master(&quot;local&quot;)
        .appName(getClass.getSimpleName.replace(&quot;$&quot;, &quot;&quot;))
        .getOrCreate()
        
      spark.sparkContext.setLogLevel(&quot;ERROR&quot;)
    }
      
  </code></pre>
  <p>And then using this in a trait to set up some sample data:</p>
  <pre class="lang-scala prettyprint-override"><code> trait TestData extends SparkSetup {
  
      def data(): DataFrame = {
  
        val testDataStruct = StructType(List(
                                StructField(&quot;date&quot;, StringType, true),
                                StructField(&quot;period&quot;, StringType, true),
                                StructField(&quot;ID&quot;, IntegerType, true),
                                StructField(&quot;SomeText&quot;, StringType, true)))
  
        val testData = Seq(Row(&quot;01012020&quot;, &quot;10:00&quot;, 20, &quot;Some Text&quot;))
  
        spark.createDataFrame(spark.sparkContext.parallelize(testData), testDataStruct)
        
      }
    }
  </code></pre>
  <p>I'm then putting these together to run the test via <code>withFixture</code> and use <code>afterAll</code> to close the SparkSession; this is clearly where something isn't quite right, but I'm not really sure what:</p>
  <pre class="lang-scala prettyprint-override"><code>  trait DataFixture extends funsuite.FixtureAnyFunSuite with TestData with BeforeAndAfterAll { this: FixtureSuite =&gt;
  
      type FixtureParam = DataFrame
  
      def withFixture(test: OneArgTest) = {
  
        super.withFixture(test.toNoArgTest(data())) // &quot;loan&quot; the fixture to the test
      }
  
      override def afterAll() {
        spark.close()
      }        
    }
  }
  </code></pre>
  <p>I'm currently testing a basic function to hash the columns in a DataFrame dynamically, with the option to exclude some; here's the code:</p>
  <pre class="lang-scala prettyprint-override"><code>package com.example.utilities
  
  import org.apache.spark.sql.SparkSession
  import org.apache.spark.sql.DataFrame
  import org.apache.spark.sql.functions.col
  import org.apache.spark.sql.functions._
  
  object GeneralTransforms {
      def addHashColumn(inputDataFrame: DataFrame, exclusionCols: List[String]): DataFrame = {
          
          val columnsToHash = inputDataFrame.columns.filterNot(exclusionCols.contains(_)).toList
          
          inputDataFrame.withColumn(&quot;RowHash&quot;, sha2(concat_ws(&quot;|&quot;, columnsToHash.map(col) : _*), 256))
      }
  }
  </code></pre>
  <p>And the current test cases:-</p>
  <pre class="lang-scala prettyprint-override"><code>import testSetup._
  import com.example.utilities.GeneralTransforms._
  
  import org.apache.spark.sql.DataFrame
  
  class TestData extends funsuite.FixtureAnyFunSuite with DataFixture {
    test(&quot;Test data has correct columns&quot;) { inputData =&gt; 
      val cols = inputData.columns.toSeq
      val expectedCols = Array(&quot;date&quot;, &quot;period&quot;, &quot;ID&quot;, &quot;SomeText&quot;).toSeq
      
      assert(cols == expectedCols)
    }
  }
  
  class TestAddHashColumn extends funsuite.FixtureAnyFunSuite with DataFixture {
    
    test(&quot;Test new hash column added&quot;) { inputData =&gt;
      val hashedDf = addHashColumn(inputData, List())
      val initialCols = inputData.columns.toSeq
      val cols = hashedDf.columns.toSeq
      
      assert(initialCols.contains(&quot;RowHash&quot;) == false)
      assert(cols.contains(&quot;RowHash&quot;) == true)
    }
  
    test(&quot;Test all columns hashed - no exclusion&quot;) { inputData =&gt;
      val hashedDf = addHashColumn(inputData, List())
      val rowHashColumn = hashedDf.select(&quot;RowHash&quot;).first().getString(0)
      val checkString = &quot;01012020|10:00|20|Some Text&quot;
      val expectedHash = String.format(&quot;%064x&quot;, new java.math.BigInteger(1, java.security.MessageDigest.getInstance(&quot;SHA-256&quot;).digest(checkString.getBytes(&quot;UTF-8&quot;))))
  
      assert(rowHashColumn == expectedHash)
    }
  
    test(&quot;Test all columns hashed - with exclusion&quot;) { inputData =&gt;
  
      val excludedColumns = List(&quot;ID&quot;, &quot;SomeText&quot;)
      val hashedDf = addHashColumn(inputData,excludedColumns)
      val rowHashColumn = hashedDf.select(&quot;RowHash&quot;).first().getString(0)
      val checkString = &quot;01012020|10:00&quot;
      val expectedHash = String.format(&quot;%064x&quot;, new java.math.BigInteger(1, java.security.MessageDigest.getInstance(&quot;SHA-256&quot;).digest(checkString.getBytes(&quot;UTF-8&quot;))))
  
      assert(rowHashColumn == expectedHash)
  
    }
  }
  </code></pre>
  <p>Both test suites work absolutely fine in isolation; it's only when running both together that I have an issue.  This can also be resolved by adding <code>parallelExecution in Test := false</code> to my build.sbt, but it would be nice to be able to allow this to happen in parallel as I add more and more tests.</p>
  <p>I also wondered if it's something that can just be resolved by running something in the BeforeAll/AfterAll that checks for other instances of SparkSession against the context, but I'm not sure how to do that and wanted to exhaust this avenue first before I go down another rabbit hole!</p>
  <h1>Edit</h1>
  <p>Since posting, I've spent some more time on this and have switched things around a bit, with a helper class for dealing with the Spark setup.  In this I've set it to create a pseudo master spark session using the <code>SparkSession.builder.getOrCreate</code> method, but then created a new spark session for the actual tests - this will allow me to have different configs, and do stuff like have different temp table registrations and so on.  However, I've still not been able to solve the shut down of spark - obviously if I run spark.stop() against any of the running sessions on the <code>SparkContext</code> it will stop the context for all sessions.</p>
  <p>And it would appear that until sbt is exited the context is not stopped?</p>