code: def test_get_or_create_and_get_active_or_create(self):
      inputd = tempfile.mkdtemp()
      outputd = tempfile.mkdtemp() + "/"
  
      def updater(vs, s):
          return sum(vs, s or 0)
  
      def setup():
          conf = SparkConf().set("spark.default.parallelism", 1)
          sc = SparkContext(conf=conf)
          ssc = StreamingContext(sc, 2)
          dstream = ssc.textFileStream(inputd).map(lambda x: (x, 1))
          wc = dstream.updateStateByKey(updater)
          wc.map(lambda x: "%s,%d" % x).saveAsTextFiles(
              outputd + "test"
          )
          wc.checkpoint(2)
          self.setupCalled = True
          return ssc
  
      # Verify that getOrCreate() calls setup() in absence of checkpoint files
      self.cpd = tempfile.mkdtemp("test_streaming_cps")
      self.setupCalled = False
      self.ssc = StreamingContext.getOrCreate(self.cpd, setup)
      self.assertTrue(self.setupCalled)
  
      self.ssc.start()
  
      def check_output(n):
          while not os.listdir(outputd):
              if self.ssc.awaitTerminationOrTimeout(0.5):
                  raise RuntimeError("ssc stopped")
          time.sleep(
              1
          )  # make sure mtime is larger than the previous one
          with open(os.path.join(inputd, str(n)), "w") as f:
              f.writelines(["%d\n" % i for i in range(10)])
  
          while True:
              if self.ssc.awaitTerminationOrTimeout(0.5):
                  raise RuntimeError("ssc stopped")
              p = os.path.join(outputd, max(os.listdir(outputd)))
              if "_SUCCESS" not in os.listdir(p):
                  # not finished
                  continue
              ordd = self.ssc.sparkContext.textFile(p).map(
                  lambda line: line.split(",")
              )
              d = ordd.values().map(int).collect()
              if not d:
                  continue
              self.assertEqual(10, len(d))
              s = set(d)
              self.assertEqual(1, len(s))
              m = s.pop()
              if n > m:
                  continue
              self.assertEqual(n, m)
              break
  
      check_output(1)
      check_output(2)
  
      # Verify the getOrCreate() recovers from checkpoint files
      self.ssc.stop(True, True)
      time.sleep(1)
      self.setupCalled = False
      self.ssc = StreamingContext.getOrCreate(self.cpd, setup)
      self.assertFalse(self.setupCalled)
      self.ssc.start()
      check_output(3)
  
      # Verify that getOrCreate() uses existing SparkContext
      self.ssc.stop(True, True)
      time.sleep(1)
      self.sc = SparkContext(conf=SparkConf())
      self.setupCalled = False
      self.ssc = StreamingContext.getOrCreate(self.cpd, setup)
      self.assertFalse(self.setupCalled)
      self.assertTrue(self.ssc.sparkContext == self.sc)
  
      # Verify the getActiveOrCreate() recovers from checkpoint files
      self.ssc.stop(True, True)
      time.sleep(1)
      self.setupCalled = False
      self.ssc = StreamingContext.getActiveOrCreate(self.cpd, setup)
      self.assertFalse(self.setupCalled)
      self.ssc.start()
      check_output(4)
  
      # Verify that getActiveOrCreate() returns active context
      self.setupCalled = False
      self.assertEqual(
          StreamingContext.getActiveOrCreate(self.cpd, setup), self.ssc
      )
      self.assertFalse(self.setupCalled)
  
      # Verify that getActiveOrCreate() uses existing SparkContext
      self.ssc.stop(True, True)
      time.sleep(1)
      self.sc = SparkContext(conf=SparkConf())
      self.setupCalled = False
      self.ssc = StreamingContext.getActiveOrCreate(self.cpd, setup)
      self.assertFalse(self.setupCalled)
      self.assertTrue(self.ssc.sparkContext == self.sc)
  
      # Verify that getActiveOrCreate() calls setup() in absence of checkpoint files
      self.ssc.stop(True, True)
      shutil.rmtree(self.cpd)  # delete checkpoint directory
      time.sleep(1)
      self.setupCalled = False
      self.ssc = StreamingContext.getActiveOrCreate(self.cpd, setup)
      self.assertTrue(self.setupCalled)
  
      # Stop everything
      self.ssc.stop(True, True)
  
imports: import operator
  import os
  import shutil
  import tempfile
  import time
  import unittest
  from functools import reduce
  from itertools import chain
  import platform
  from pyspark import SparkConf, SparkContext
  from pyspark.streaming import StreamingContext
  from pyspark.testing.streamingutils import PySparkStreamingTestCase
label_desc: Use image node_id 5 for calling the CheckpointTests obj's underlying member method code with example usage: obj.test_get_or_create_and_get_active_or_create() and returns: sum, ssc