class_docstr: A collection of data which can be pre-processed in parallel on Spark.
class_docstr_tok: ["A","collection","of","data","which","can","be","pre-processed","in","parallel","on","Spark","."]
code: def to_spark_df(self) -> "SparkDataFrame":
      if self._get_class_name() != "pandas.core.frame.DataFrame":
          invalidInputError(
              False,
              "Currently only support to_spark_df on XShards of Pandas DataFrame",
          )
  
      try:
          import pyarrow as pa
  
          sdf_schema = self._get_spark_df_schema()
  
          sqlContext = get_spark_sql_context(get_spark_context())
          timezone = sqlContext._conf.sessionLocalTimeZone()
  
          def f(iter):
              from bigdl.dllib.utils.log4Error import invalidInputError
  
              pdf_list = list(iter)
              invalidInputError(
                  len(pdf_list) == 1,
                  f"For XShards of pandas dataframe, expects there is only 1"
                  f" pandas dataframe for each partition, but got {len(pdf_list)}",
              )
              for pdf in pdf_list:
                  import os
                  import uuid
                  from pyspark.sql.pandas.types import to_arrow_type
                  from pyspark.sql.pandas.serializers import (
                      ArrowStreamPandasSerializer,
                  )
                  from tempfile import NamedTemporaryFile
  
                  tmpFile = "/tmp/" + str(uuid.uuid1())
                  os.mkdir(tmpFile)
  
                  arrow_types = [
                      to_arrow_type(f.dataType)
                      for f in sdf_schema.fields
                  ]
                  arrow_data = [
                      [
                          (c, t)
                          for (_, c), t in zip(pdf.items(), arrow_types)
                      ]
                  ]
                  col_by_name = True
                  safecheck = False
                  ser = ArrowStreamPandasSerializer(
                      timezone, safecheck, col_by_name
                  )
  
                  tempFile = NamedTemporaryFile(
                      delete=False, dir=tmpFile
                  )
                  try:
                      ser.dump_stream(arrow_data, tempFile)
                  finally:
                      tempFile.close()
                  return [tempFile.name]
  
          jiter = self.rdd.mapPartitions(f)
          from bigdl.dllib.utils.file_utils import callZooFunc
  
          df = callZooFunc(
              "float",
              "orcaToDataFrame",
              jiter,
              sdf_schema.json(),
              sqlContext,
          )
          return df
      except Exception as e:
          print(
              f"createDataFrame from shards attempted Arrow optimization failed as: {str(e)},"
              f"Will try without Arrow optimization"
          )
          return self._to_spark_df_without_arrow()
  
imports: from py4j.protocol import Py4JError
  from bigdl.orca.data.utils import *
  from bigdl.orca import OrcaContext
  from bigdl.dllib.utils.common import get_node_and_core_number, get_spark_sql_context, get_spark_context
  from bigdl.dllib.utils import nest
  from bigdl.dllib.utils.log4Error import invalidInputError
  import numpy
  from typing import TYPE_CHECKING, Any
  from typing import Callable, Dict, List, Optional, Tuple, Union
label_desc: Use image node_id 28 for calling the SparkXShards obj's underlying member method code with example usage: obj.to_spark_df() and returns: df, self, list