code: def spark_df_to_rdd_pd(
      df: "DataFrame",
      squeeze: bool = False,
      index_col: Optional[str] = None,
      dtype: Optional[
          Union[
              str,
              Dict[str, str],
              Dict[str, Type[float32]],
              Dict[int, Union[Type[float32], Type[int32]]],
          ]
      ] = None,
      index_map: Optional[Dict[int, str]] = None,
  ) -> "RDD":
      from bigdl.orca import OrcaContext
  
      import pyspark.sql.functions as F
      import pyspark.sql.types as T
  
      to_array = F.udf(
          lambda v: v.toArray().tolist(), T.ArrayType(T.FloatType())
      )
      for colName, colType in df.dtypes:
          if colType == "vector":
              df = df.withColumn(colName, to_array(colName))
  
      shard_size = OrcaContext._shard_size
  
      try:
          import pyarrow as pa
  
          pd_rdd = to_pandas(
              df,
              squeeze,
              index_col,
              dtype,
              index_map,
              batch_size=shard_size,
          )
          return pd_rdd
      except Exception as e:
          print(
              f"create shards from Spark DataFrame attempted Arrow optimization failed as:"
              f" {str(e)}. Will try without Arrow optimization"
          )
          pd_rdd = df.rdd.mapPartitions(
              to_pandas_without_arrow(
                  df.columns,
                  squeeze,
                  index_col,
                  dtype,
                  index_map,
                  batch_size=shard_size,
              )
          )
          return pd_rdd
  
imports: from bigdl.dllib.utils.file_utils import get_file_list, callZooFunc
  from bigdl.dllib.utils.utils import convert_row_to_numpy
  from bigdl.dllib.utils.common import *
  from bigdl.dllib.utils.log4Error import *
  import pyspark.sql.functions
  from typing import TYPE_CHECKING, Dict, List, Optional, Type, Union
  from numpy import float32, int32
label_desc: Use image node_id 20 for calling a global function with example usage: spark_df_to_rdd_pd(df, squeeze, index_col, dtype, index_map) and returns: pd_rdd, pd_rdd