code: def make_spark_converter(
      df,
      parquet_row_group_size_bytes=DEFAULT_ROW_GROUP_SIZE_BYTES,
      compression_codec=None,
      dtype="float32",
  ):
      """Convert a spark dataframe into a :class:`SparkDatasetConverter` object.
      It will materialize a spark dataframe to the directory specified by
      spark conf 'petastorm.spark.converter.parentCacheDirUrl'.
      The dataframe will be materialized in parquet format, and we can specify
      `parquet_row_group_size_bytes` and `compression_codec` for the parquet
      format. See params documentation for details.
  
      The returned `SparkDatasetConverter` object will hold the materialized
      dataframe, and can be used to make one or more tensorflow datasets or
      torch dataloaders.
  
      We can explicitly delete the materialized dataframe data, see
      `SparkDatasetConverter.delete`, and when the spark application exit,
      it will try best effort to delete the materialized dataframe data.
  
      :param df: The :class:`pyspark.sql.DataFrame` object to be converted,
                 or a string of path pointing to the directory that stores the dataframe data
                 as parquet format, on databricks runtime, the path must be a dbfs
                 fuse path like 'file:/dbfs/xxx' or a dbfs path like 'dbfs:/xxx'.
      :param parquet_row_group_size_bytes: An int denoting the number of bytes
          in a parquet row group when materializing the dataframe.
      :param compression_codec: Specify compression codec.
          It can be one of 'uncompressed', 'bzip2', 'gzip', 'lz4', 'snappy', 'deflate'.
          Default ``None``. If ``None``, it will leave the data uncompressed.
      :param dtype: ``None``, 'float32' or 'float64', specifying the precision of the floating-point
          elements in the output dataset. Integer types will remain unchanged. If ``None``, all types
          will remain unchanged. Default 'float32'.
  
      :return: a :class:`SparkDatasetConverter` object that holds the
          materialized dataframe and can be used to make one or more tensorflow
          datasets or torch dataloaders.
      """
  
      parent_cache_dir_url = _get_parent_cache_dir_url()
  
      if isinstance(df, str):
          dataset_dir_url = df
          if "DATABRICKS_RUNTIME_VERSION" in os.environ:
              dataset_dir_url = _normalize_databricks_dbfs_url(
                  dataset_dir_url,
                  "On databricks runtime, if `df` argument is a string, it must be a dbfs "
                  "fuse path like 'file:/dbfs/xxx' or a dbfs path like 'dbfs:/xxx'.",
              )
      else:
          # TODO: Improve default behavior to be automatically choosing the best way.
          compression_codec = compression_codec or "uncompressed"
  
          if compression_codec.lower() not in [
              "uncompressed",
              "bzip2",
              "gzip",
              "lz4",
              "snappy",
              "deflate",
          ]:
              raise RuntimeError(
                  "compression_codec should be None or one of the following values: "
                  "'uncompressed', 'bzip2', 'gzip', 'lz4', 'snappy', 'deflate'"
              )
  
          dataset_dir_url = _cache_df_or_retrieve_cache_data_url(
              df,
              parent_cache_dir_url,
              parquet_row_group_size_bytes,
              compression_codec,
              dtype,
          )
  
      # TODO: improve this by read parquet file metadata to get count
      #  Currently spark can make sure to only read the minimal column
      #  so count will usually be fast.
      spark = _get_spark_session()
      spark_df = spark.read.parquet(dataset_dir_url)
  
      dataset_size = spark_df.count()
      parquet_file_url_list = list(spark_df._jdf.inputFiles())
      _check_dataset_file_median_size(parquet_file_url_list)
  
      return SparkDatasetConverter(
          dataset_dir_url, parquet_file_url_list, dataset_size
      )
  
imports: import atexit
  import datetime
  import logging
  import os
  import shutil
  import threading
  import time
  import uuid
  from distutils.version import LooseVersion
  from multiprocessing.pool import ThreadPool
  from typing import List, Any
  import pyspark
  from pyarrow import LocalFileSystem
  from pyspark.sql.session import SparkSession
  from pyspark.sql.types import ArrayType, DoubleType, FloatType
  from six.moves.urllib.parse import urlparse
  from petastorm import make_batch_reader
  from petastorm.fs_utils import FilesystemResolver, get_filesystem_and_path_or_paths, normalize_dir_url
  from fsspec.core import strip_protocol
label_desc: Use image node_id 22 for calling a global function with example usage: make_spark_converter(df, parquet_row_group_size_bytes, compression_codec, dtype) and returns: SparkDatasetConverter