code: def test_streaming_foreach_batch_tempview(self):
      q = None
  
      def collectBatch(batch_df, batch_id):
          batch_df.createOrReplaceTempView("updates")
          # it should use the spark session within given DataFrame, as microbatch execution will
          # clone the session which is no longer same with the session used to start the
          # streaming query
          assert (
              len(
                  batch_df.sparkSession.sql(
                      "SELECT * FROM updates"
                  ).collect()
              )
              == 2
          )
          # Write to a global view verify on the repl/client side.
          batch_df.createOrReplaceGlobalTempView("temp_view")
  
      try:
          df = self.spark.readStream.format("text").load(
              "python/test_support/sql/streaming"
          )
          q = df.writeStream.foreachBatch(collectBatch).start()
          q.processAllAvailable()
          collected = self.spark.sql(
              "SELECT * FROM global_temp.temp_view"
          ).collect()
          self.assertTrue(len(collected[0]), 2)
      finally:
          if q:
              q.stop()
  
file_path: spark/python/pyspark/sql/tests/streaming/test_streaming_foreach_batch.py
imports: import time
  from pyspark.sql.dataframe import DataFrame
  from pyspark.testing.sqlutils import ReusedSQLTestCase
label_desc: Use image node_id 2 for calling the StreamingTestsForeachBatchMixin obj's underlying member method code with example usage: obj.test_streaming_foreach_batch_tempview() without return types