code: def createDataFrame(
      self,
      data: Union["pd.DataFrame", "np.ndarray", Iterable[Any]],
      schema: Optional[
          Union[AtomicType, StructType, str, List[str], Tuple[str, ...]]
      ] = None,
  ) -> "DataFrame":
      assert data is not None
      if isinstance(data, DataFrame):
          raise PySparkTypeError(
              error_class="INVALID_TYPE",
              message_parameters={
                  "arg_name": "data",
                  "data_type": "DataFrame",
              },
          )
  
      _schema: Optional[Union[AtomicType, StructType]] = None
      _cols: Optional[List[str]] = None
      _num_cols: Optional[int] = None
  
      if isinstance(schema, str):
          schema = self.client._analyze(  # type: ignore[assignment]
              method="ddl_parse", ddl_string=schema
          ).parsed
  
      if isinstance(schema, (AtomicType, StructType)):
          _schema = schema
          if isinstance(schema, StructType):
              _num_cols = len(schema.fields)
          else:
              _num_cols = 1
  
      elif isinstance(schema, (list, tuple)):
          # Must re-encode any unicode strings to be consistent with StructField names
          _cols = [
              x.encode("utf-8") if not isinstance(x, str) else x
              for x in schema
          ]
          _num_cols = len(_cols)
  
      elif schema is not None:
          raise PySparkTypeError(
              error_class="NOT_LIST_OR_NONE_OR_STRUCT",
              message_parameters={
                  "arg_name": "schema",
                  "arg_type": type(schema).__name__,
              },
          )
  
      if isinstance(data, np.ndarray) and data.ndim not in [1, 2]:
          raise PySparkValueError(
              error_class="INVALID_NDARRAY_DIMENSION",
              message_parameters={"dimensions": "1 or 2"},
          )
      elif isinstance(data, Sized) and len(data) == 0:
          if _schema is not None:
              return DataFrame(
                  LocalRelation(table=None, schema=_schema.json()), self
              )
          else:
              raise PySparkValueError(
                  error_class="CANNOT_INFER_EMPTY_SCHEMA",
                  message_parameters={},
              )
  
      _table: Optional[pa.Table] = None
  
      if isinstance(data, pd.DataFrame):
          # Logic was borrowed from `_create_from_pandas_with_arrow` in
          # `pyspark.sql.pandas.conversion.py`. Should ideally deduplicate the logics.
  
          # If no schema supplied by user then get the names of columns only
          if schema is None:
              _cols = [
                  str(x) if not isinstance(x, str) else x
                  for x in data.columns
              ]
          elif isinstance(schema, (list, tuple)) and cast(
              int, _num_cols
          ) < len(data.columns):
              assert isinstance(_cols, list)
              _cols.extend(
                  [
                      f"_{i + 1}"
                      for i in range(
                          cast(int, _num_cols), len(data.columns)
                      )
                  ]
              )
              _num_cols = len(_cols)
  
          # Determine arrow types to coerce data when creating batches
          arrow_schema: Optional[pa.Schema] = None
          spark_types: List[Optional[DataType]]
          arrow_types: List[Optional[pa.DataType]]
          if isinstance(schema, StructType):
              deduped_schema = cast(
                  StructType, _deduplicate_field_names(schema)
              )
              spark_types = [
                  field.dataType for field in deduped_schema.fields
              ]
              arrow_schema = to_arrow_schema(deduped_schema)
              arrow_types = [field.type for field in arrow_schema]
              _cols = [
                  str(x) if not isinstance(x, str) else x
                  for x in schema.fieldNames()
              ]
          elif isinstance(schema, DataType):
              raise PySparkTypeError(
                  error_class="UNSUPPORTED_DATA_TYPE_FOR_ARROW",
                  message_parameters={"data_type": str(schema)},
              )
          else:
              # Any timestamps must be coerced to be compatible with Spark
              spark_types = [
                  TimestampType()
                  if is_datetime64_dtype(t)
                  or isinstance(t, pd.DatetimeTZDtype)
                  else DayTimeIntervalType()
                  if is_timedelta64_dtype(t)
                  else None
                  for t in data.dtypes
              ]
              arrow_types = [
                  to_arrow_type(dt) if dt is not None else None
                  for dt in spark_types
              ]
  
          timezone, safecheck = self._client.get_configs(
              "spark.sql.session.timeZone",
              "spark.sql.execution.pandas.convertToArrowArraySafely",
          )
  
          ser = ArrowStreamPandasSerializer(
              cast(str, timezone), safecheck == "true"
          )
  
          _table = pa.Table.from_batches(
              [
                  ser._create_batch(
                      [
                          (c, at, st)
                          for (_, c), at, st in zip(
                              data.items(), arrow_types, spark_types
                          )
                      ]
                  )
              ]
          )
  
          if isinstance(schema, StructType):
              assert arrow_schema is not None
              _table = _table.rename_columns(
                  cast(
                      StructType, _deduplicate_field_names(schema)
                  ).names
              ).cast(arrow_schema)
  
      elif isinstance(data, np.ndarray):
          if _cols is None:
              if data.ndim == 1 or data.shape[1] == 1:
                  _cols = ["value"]
              else:
                  _cols = [
                      "_%s" % i for i in range(1, data.shape[1] + 1)
                  ]
  
          if data.ndim == 1:
              if 1 != len(_cols):
                  raise PySparkValueError(
                      error_class="AXIS_LENGTH_MISMATCH",
                      message_parameters={
                          "expected_length": str(len(_cols)),
                          "actual_length": "1",
                      },
                  )
  
              _table = pa.Table.from_arrays([pa.array(data)], _cols)
          else:
              if data.shape[1] != len(_cols):
                  raise PySparkValueError(
                      error_class="AXIS_LENGTH_MISMATCH",
                      message_parameters={
                          "expected_length": str(len(_cols)),
                          "actual_length": str(data.shape[1]),
                      },
                  )
  
              _table = pa.Table.from_arrays(
                  [
                      pa.array(data[::, i])
                      for i in range(0, data.shape[1])
                  ],
                  _cols,
              )
  
          # The _table should already have the proper column names.
          _cols = None
  
      else:
          _data = list(data)
  
          if isinstance(_data[0], dict):
              # Sort the data to respect inferred schema.
              # For dictionaries, we sort the schema in alphabetical order.
              _data = [
                  dict(sorted(d.items())) if d is not None else None
                  for d in _data
              ]
  
          elif not isinstance(
              _data[0], (Row, tuple, list, dict)
          ) and not hasattr(_data[0], "__dict__"):
              # input data can be [1, 2, 3]
              # we need to convert it to [[1], [2], [3]] to be able to infer schema.
              _data = [[d] for d in _data]
  
          if _schema is not None:
              if not isinstance(_schema, StructType):
                  _schema = StructType().add("value", _schema)
          else:
              _schema = self._inferSchemaFromList(_data, _cols)
  
              if _cols is not None and cast(int, _num_cols) < len(
                  _cols
              ):
                  _num_cols = len(_cols)
  
              if _has_nulltype(_schema):
                  # For cases like createDataFrame([("Alice", None, 80.1)], schema)
                  # we can not infer the schema from the data itself.
                  raise PySparkValueError(
                      error_class="CANNOT_DETERMINE_TYPE",
                      message_parameters={},
                  )
  
          from pyspark.sql.connect.conversion import (
              LocalDataToArrowConversion,
          )
  
          # Spark Connect will try its best to build the Arrow table with the
          # inferred schema in the client side, and then rename the columns and
          # cast the datatypes in the server side.
          _table = LocalDataToArrowConversion.convert(_data, _schema)
  
      # TODO: Beside the validation on number of columns, we should also check
      # whether the Arrow Schema is compatible with the user provided Schema.
      if _num_cols is not None and _num_cols != _table.shape[1]:
          raise PySparkValueError(
              error_class="AXIS_LENGTH_MISMATCH",
              message_parameters={
                  "expected_length": str(_num_cols),
                  "actual_length": str(_table.shape[1]),
              },
          )
  
      if _schema is not None:
          local_relation = LocalRelation(_table, schema=_schema.json())
      else:
          local_relation = LocalRelation(_table)
  
      cache_threshold = self._client.get_configs(
          "spark.sql.session.localRelationCacheThreshold"
      )
      plan: LogicalPlan = local_relation
      if (
          cache_threshold[0] is not None
          and int(cache_threshold[0]) <= _table.nbytes
      ):
          plan = CachedLocalRelation(
              self._cache_local_relation(local_relation)
          )
  
      df = DataFrame(plan, self)
      if _cols is not None and len(_cols) > 0:
          df = df.toDF(*_cols)
      return df
  
imports: from pyspark.sql.connect.utils import check_dependencies
  import threading
  import os
  import warnings
  from collections.abc import Sized
  from functools import reduce
  from threading import RLock
  from typing import Optional, Any, Union, Dict, List, Tuple, Set, cast, overload, Iterable, TYPE_CHECKING, ClassVar
  import numpy
  import pandas
  import pyarrow
  from pandas.api.types import is_datetime64_dtype, is_timedelta64_dtype
  import urllib
  from pyspark import SparkContext, SparkConf, __version__
  from pyspark.loose_version import LooseVersion
  from pyspark.sql.connect.client import SparkConnectClient, ChannelBuilder
  from pyspark.sql.connect.conf import RuntimeConf
  from pyspark.sql.connect.dataframe import DataFrame
  from pyspark.sql.connect.plan import SQL, Range, LocalRelation, LogicalPlan, CachedLocalRelation, CachedRelation, CachedRemoteRelation
  from pyspark.sql.connect.readwriter import DataFrameReader
  from pyspark.sql.connect.streaming.readwriter import DataStreamReader
  from pyspark.sql.connect.streaming.query import StreamingQueryManager
  from pyspark.sql.pandas.serializers import ArrowStreamPandasSerializer
  from pyspark.sql.pandas.types import to_arrow_schema, to_arrow_type, _deduplicate_field_names
  from pyspark.sql.session import classproperty, SparkSession
  from pyspark.sql.types import _infer_schema, _has_nulltype, _merge_type, Row, DataType, DayTimeIntervalType, StructType, AtomicType, TimestampType
  from pyspark.sql.utils import to_str
  from pyspark.errors import PySparkAttributeError, PySparkNotImplementedError, PySparkRuntimeError, PySparkValueError, PySparkTypeError
label_desc: Use image node_id 10 for calling the SparkSession obj's underlying member method code with example usage: obj.createDataFrame(data, schema) and returns: df, DataFrame