code: def __init__(
      self,
      idempotent,
      sc,
      spark_conf,
      app_name,
      master,
      local,
      log,
      quiet,
      append,
      min_block_size,
      branching_factor,
      tmpdir,
      local_tmpdir,
      skip_logging_configuration,
      optimizer_iterations,
      *,
      gcs_requester_pays_project: Optional[str] = None,
      gcs_requester_pays_buckets: Optional[str] = None,
  ):
      assert (
          gcs_requester_pays_project is not None
          or gcs_requester_pays_buckets is None
      )
  
      try:
          local_jar_info = local_jar_information()
      except ValueError:
          local_jar_info = None
  
      if local_jar_info is not None:
          conf = pyspark.SparkConf()
  
          base_conf = spark_conf or {}
          for k, v in base_conf.items():
              conf.set(k, v)
  
          jars = [local_jar_info.path]
          extra_classpath = local_jar_info.extra_classpath
  
          if (
              os.environ.get("HAIL_SPARK_MONITOR")
              or os.environ.get("AZURE_SPARK") == "1"
          ):
              import sparkmonitor
  
              jars.append(
                  os.path.join(
                      os.path.dirname(sparkmonitor.__file__),
                      "listener.jar",
                  )
              )
              append_to_comma_separated_list(
                  conf,
                  "spark.extraListeners",
                  "sparkmonitor.listener.JupyterSparkMonitorListener",
              )
  
          append_to_comma_separated_list(conf, "spark.jars", *jars)
          if os.environ.get("AZURE_SPARK") == "1":
              print(
                  'AZURE_SPARK environment variable is set to "1", assuming you are in HDInsight.'
              )
              # Setting extraClassPath in HDInsight overrides the classpath entirely so you can't
              # load the Scala standard library. Interestingly, setting extraClassPath is not
              # necessary in HDInsight.
          else:
              append_to_comma_separated_list(
                  conf,
                  "spark.driver.extraClassPath",
                  *jars,
                  *extra_classpath,
              )
              append_to_comma_separated_list(
                  conf,
                  "spark.executor.extraClassPath",
                  "./hail-all-spark.jar",
                  *extra_classpath,
              )
  
          if sc is None:
              pyspark.SparkContext._ensure_initialized(conf=conf)
          elif not quiet:
              sys.stderr.write(
                  "pip-installed Hail requires additional configuration options in Spark referring\n"
                  "  to the path to the Hail Python module directory HAIL_DIR,\n"
                  "  e.g. /path/to/python/site-packages/hail:\n"
                  "    spark.jars=HAIL_DIR/backend/hail-all-spark.jar\n"
                  "    spark.driver.extraClassPath=HAIL_DIR/backend/hail-all-spark.jar\n"
                  "    spark.executor.extraClassPath=./hail-all-spark.jar"
              )
      else:
          pyspark.SparkContext._ensure_initialized()
  
      self._gateway = pyspark.SparkContext._gateway
      jvm = pyspark.SparkContext._jvm
      assert jvm
  
      hail_package = getattr(jvm, "is").hail
      jsc = sc._jsc.sc() if sc else None
  
      if idempotent:
          jbackend = (
              hail_package.backend.spark.SparkBackend.getOrCreate(
                  jsc,
                  app_name,
                  master,
                  local,
                  log,
                  True,
                  append,
                  skip_logging_configuration,
                  min_block_size,
                  tmpdir,
                  local_tmpdir,
                  gcs_requester_pays_project,
                  gcs_requester_pays_buckets,
              )
          )
          jhc = hail_package.HailContext.getOrCreate(
              jbackend, branching_factor, optimizer_iterations
          )
      else:
          jbackend = hail_package.backend.spark.SparkBackend.apply(
              jsc,
              app_name,
              master,
              local,
              log,
              True,
              append,
              skip_logging_configuration,
              min_block_size,
              tmpdir,
              local_tmpdir,
              gcs_requester_pays_project,
              gcs_requester_pays_buckets,
          )
          jhc = hail_package.HailContext.apply(
              jbackend, branching_factor, optimizer_iterations
          )
  
      self._jsc = jbackend.sc()
      if sc:
          self.sc = sc
      else:
          self.sc = pyspark.SparkContext(
              gateway=self._gateway, jsc=jvm.JavaSparkContext(self._jsc)
          )
      self._jspark_session = jbackend.sparkSession()
      self._spark_session = pyspark.sql.SparkSession(
          self.sc, self._jspark_session
      )
  
      super(SparkBackend, self).__init__(jvm, jbackend, jhc)
  
      self._fs = None
      self._logger = None
  
      if not quiet:
          sys.stderr.write(
              "Running on Apache Spark version {}\n".format(
                  self.sc.version
              )
          )
          if self._jsc.uiWebUrl().isDefined():
              sys.stderr.write(
                  "SparkUI available at {}\n".format(
                      self._jsc.uiWebUrl().get()
                  )
              )
  
          jbackend.startProgressBar()
  
      self._initialize_flags({})
  
      self._router_async_fs = RouterAsyncFS(
          gcs_kwargs={
              "gcs_requester_pays_configuration": gcs_requester_pays_project
          }
      )
  
imports: import sys
  import os
  import pyspark
  import pyspark.sql
  import orjson
  from typing import Optional
  from hail.expr.table_type import ttable
  from hail.fs.hadoop_fs import HadoopFS
  from hail.ir.renderer import CSERenderer
  from hail.table import Table
  from hailtop.aiotools.router_fs import RouterAsyncFS
  from hailtop.aiotools.validators import validate_file
  from .py4j_backend import Py4JBackend
  from .backend import local_jar_information
label_desc: Use image node_id 1 to create a new SparkBackend object from inherited base classes: Py4JBackend with example: obj = SparkBackend(idempotent, sc, spark_conf, app_name, master, local, log, quiet, append, min_block_size, branching_factor, tmpdir, local_tmpdir, skip_logging_configuration, optimizer_iterations)