code: def read_spark_io(
      path: Optional[str] = None,
      format: Optional[str] = None,
      schema: Union[str, "StructType"] = None,
      index_col: Optional[Union[str, List[str]]] = None,
      **options: Any,
  ) -> DataFrame:
      """Load a DataFrame from a Spark data source.
  
      Parameters
      ----------
      path : string, optional
          Path to the data source.
      format : string, optional
          Specifies the output data source format. Some common ones are:
  
          - 'delta'
          - 'parquet'
          - 'orc'
          - 'json'
          - 'csv'
      schema : string or StructType, optional
          Input schema. If none, Spark tries to infer the schema automatically.
          The schema can either be a Spark StructType, or a DDL-formatted string like
          `col0 INT, col1 DOUBLE`.
      index_col : str or list of str, optional, default: None
          Index column of table in Spark.
      options : dict
          All other options passed directly into Spark's data source.
  
      See Also
      --------
      DataFrame.read_table
      DataFrame.read_delta
      DataFrame.read_parquet
  
      Examples
      --------
      >>> ps.range(1).spark.to_spark_io('%s/read_spark_io/data.parquet' % path)
      >>> ps.read_spark_io(
      ...     '%s/read_spark_io/data.parquet' % path, format='parquet', schema='id long')
         id
      0   0
  
      >>> ps.range(10, 15, num_partitions=1).spark.to_spark_io('%s/read_spark_io/data.json' % path,
      ...                                                format='json', lineSep='__')
      >>> ps.read_spark_io(
      ...     '%s/read_spark_io/data.json' % path, format='json', schema='id long', lineSep='__')
         id
      0  10
      1  11
      2  12
      3  13
      4  14
  
      You can preserve the index in the roundtrip as below.
  
      >>> ps.range(10, 15, num_partitions=1).spark.to_spark_io('%s/read_spark_io/data.orc' % path,
      ...                                                format='orc', index_col="index")
      >>> ps.read_spark_io(
      ...     path=r'%s/read_spark_io/data.orc' % path, format="orc", index_col="index")
      ... # doctest: +NORMALIZE_WHITESPACE
             id
      index
      0      10
      1      11
      2      12
      3      13
      4      14
      """
      if (
          "options" in options
          and isinstance(options.get("options"), dict)
          and len(options) == 1
      ):
          options = options.get("options")
  
      sdf = default_session().read.load(
          path=path, format=format, schema=schema, **options
      )
      index_spark_columns, index_names = _get_index_map(sdf, index_col)
  
      return DataFrame(
          InternalFrame(
              spark_frame=sdf,
              index_spark_columns=index_spark_columns,
              index_names=index_names,
          )
      )
  
imports: from typing import Any, Callable, Dict, List, Optional, Set, Sized, Tuple, Type, Union, cast, no_type_check
  from collections.abc import Iterable
  from datetime import tzinfo
  from functools import reduce
  from io import BytesIO
  import json
  import warnings
  import numpy
  import pandas
  from pandas.api.types import is_datetime64_dtype, is_list_like
  from pandas.tseries.offsets import DateOffset
  import pyarrow
  import pyarrow.parquet
  from pyspark.sql import functions, Column
  from pyspark.sql.functions import pandas_udf
  from pyspark.sql.types import ByteType, ShortType, IntegerType, LongType, FloatType, DoubleType, BooleanType, TimestampType, TimestampNTZType, DecimalType, StringType, DateType, StructType, DataType
  from pyspark.sql.dataframe import DataFrame
  from pyspark import pandas
  from pyspark.pandas._typing import Axis, Dtype, Label, Name
  from pyspark.pandas.base import IndexOpsMixin
  from pyspark.pandas.utils import align_diff_frames, default_session, is_name_like_tuple, is_name_like_value, name_like_string, same_anchor, scol_for, validate_axis, log_advice
  from pyspark.pandas.frame import DataFrame, _reduce_spark_multi
  from pyspark.pandas.internal import InternalFrame, DEFAULT_SERIES_NAME, HIDDEN_COLUMNS, SPARK_INDEX_NAME_FORMAT
  from pyspark.pandas.series import Series, first_series
  from pyspark.pandas.spark.utils import as_nullable_spark_type, force_decimal_precision_scale
  from pyspark.pandas.indexes import Index, DatetimeIndex, TimedeltaIndex
  from pyspark.pandas.indexes.multi import MultiIndex
  from pyspark.sql.utils import get_column_class
label_desc: Use image node_id 7 for calling a global function with example usage: read_spark_io(path, format, schema, index_col) and returns: DataFrame