code: def _create_converter_from_pandas(
      data_type: DataType,
      *,
      timezone: Optional[str] = None,
      error_on_duplicated_field_names: bool = True,
      ignore_unexpected_complex_type_values: bool = False,
  ) -> Callable[["pd.Series"], "pd.Series"]:
      """
      Create a converter of pandas Series to create Spark DataFrame with Arrow optimization.
  
      Parameters
      ----------
      data_type : :class:`DataType`
          The data type corresponding to the pandas Series to be converted.
      timezone : str, optional
          The timezone to convert from. If there is a timestamp type, it's required.
      error_on_duplicated_field_names : bool, optional
          Whether raise an exception when there are duplicated field names.
          (default ``True``)
      ignore_unexpected_complex_type_values : bool, optional
          Whether ignore the case where unexpected values are given for complex types.
          If ``False``, each complex type expects:
  
          * array type: :class:`Iterable`
          * map type: :class:`dict`
          * struct type: :class:`dict` or :class:`tuple`
  
          and raise an AssertionError when the given value is not the expected type.
          If ``True``, just ignore and return the give value.
          (default ``False``)
  
      Returns
      -------
      The converter of `pandas.Series`
      """
      import pandas as pd
  
      if isinstance(data_type, TimestampType):
          assert timezone is not None
  
          def correct_timestamp(pser: pd.Series) -> pd.Series:
              return _check_series_convert_timestamps_internal(
                  pser, cast(str, timezone)
              )
  
          return correct_timestamp
  
      def _converter(dt: DataType) -> Optional[Callable[[Any], Any]]:
          if isinstance(dt, ArrayType):
              _element_conv = _converter(dt.elementType)
  
              if ignore_unexpected_complex_type_values:
                  if _element_conv is None:
  
                      def convert_array(value: Any) -> Any:
                          if isinstance(value, Iterable):
                              return list(value)
                          else:
                              return value
  
                  else:
  
                      def convert_array(value: Any) -> Any:
                          if isinstance(value, Iterable):
                              return [
                                  _element_conv(v) if v is not None else None  # type: ignore[misc]
                                  for v in value
                              ]
                          else:
                              return value
  
              else:
                  if _element_conv is None:
  
                      def convert_array(value: Any) -> Any:
                          return list(value)
  
                  else:
  
                      def convert_array(value: Any) -> Any:
                          # Iterable
                          return [
                              _element_conv(v) if v is not None else None  # type: ignore[misc]
                              for v in value
                          ]
  
              return convert_array
  
          elif isinstance(dt, MapType):
              _key_conv = _converter(dt.keyType)
              _value_conv = _converter(dt.valueType)
  
              if ignore_unexpected_complex_type_values:
                  if _key_conv is None and _value_conv is None:
  
                      def convert_map(value: Any) -> Any:
                          if isinstance(value, dict):
                              return list(value.items())
                          else:
                              return value
  
                  else:
  
                      def convert_map(value: Any) -> Any:
                          if isinstance(value, dict):
                              return [
                                  (
                                      _key_conv(k)
                                      if _key_conv is not None
                                      and k is not None
                                      else k,
                                      _value_conv(v)
                                      if _value_conv is not None
                                      and v is not None
                                      else v,
                                  )
                                  for k, v in value.items()
                              ]
                          else:
                              return value
  
              else:
                  if _key_conv is None and _value_conv is None:
  
                      def convert_map(value: Any) -> Any:
                          # dict
                          return list(value.items())
  
                  else:
  
                      def convert_map(value: Any) -> Any:
                          # dict
                          return [
                              (
                                  _key_conv(k)
                                  if _key_conv is not None
                                  and k is not None
                                  else k,
                                  _value_conv(v)
                                  if _value_conv is not None
                                  and v is not None
                                  else v,
                              )
                              for k, v in value.items()
                          ]
  
              return convert_map
  
          elif isinstance(dt, StructType):
              field_names = dt.names
  
              if error_on_duplicated_field_names and len(
                  set(field_names)
              ) != len(field_names):
                  raise UnsupportedOperationException(
                      error_class="DUPLICATED_FIELD_NAME_IN_ARROW_STRUCT",
                      message_parameters={
                          "field_names": str(field_names)
                      },
                  )
  
              dedup_field_names = _dedup_names(field_names)
  
              field_convs = [_converter(f.dataType) for f in dt.fields]
  
              if ignore_unexpected_complex_type_values:
                  if all(conv is None for conv in field_convs):
  
                      def convert_struct(value: Any) -> Any:
                          if isinstance(value, dict):
                              return {
                                  name: value.get(key, None)
                                  for name, key in zip(
                                      dedup_field_names, field_names
                                  )
                              }
                          elif isinstance(value, tuple):
                              return dict(zip(dedup_field_names, value))
                          else:
                              return value
  
                  else:
  
                      def convert_struct(value: Any) -> Any:
                          if isinstance(value, dict):
                              return {
                                  name: conv(v)
                                  if conv is not None and v is not None
                                  else v
                                  for name, conv, v in zip(
                                      dedup_field_names,
                                      field_convs,
                                      (
                                          value.get(key, None)
                                          for key in field_names
                                      ),
                                  )
                              }
                          elif isinstance(value, tuple):
                              return {
                                  name: conv(v)
                                  if conv is not None and v is not None
                                  else v
                                  for name, conv, v in zip(
                                      dedup_field_names,
                                      field_convs,
                                      value,
                                  )
                              }
                          else:
                              return value
  
              else:
                  if all(conv is None for conv in field_convs):
  
                      def convert_struct(value: Any) -> Any:
                          if isinstance(value, dict):
                              return {
                                  name: value.get(key, None)
                                  for name, key in zip(
                                      dedup_field_names, field_names
                                  )
                              }
                          else:
                              # tuple
                              return dict(zip(dedup_field_names, value))
  
                  else:
  
                      def convert_struct(value: Any) -> Any:
                          if isinstance(value, dict):
                              return {
                                  name: conv(v)
                                  if conv is not None and v is not None
                                  else v
                                  for name, conv, v in zip(
                                      dedup_field_names,
                                      field_convs,
                                      (
                                          value.get(key, None)
                                          for key in field_names
                                      ),
                                  )
                              }
                          else:
                              # tuple
                              return {
                                  name: conv(v)
                                  if conv is not None and v is not None
                                  else v
                                  for name, conv, v in zip(
                                      dedup_field_names,
                                      field_convs,
                                      value,
                                  )
                              }
  
              return convert_struct
  
          elif isinstance(dt, TimestampType):
              assert timezone is not None
  
              def convert_timestamp(value: Any) -> Any:
                  if (
                      isinstance(value, datetime.datetime)
                      and value.tzinfo is not None
                  ):
                      ts = pd.Timstamp(value)
                  else:
                      ts = pd.Timestamp(value).tz_localize(timezone)
                  return ts.to_pydatetime()
  
              return convert_timestamp
  
          elif isinstance(dt, UserDefinedType):
              udt: UserDefinedType = dt
  
              conv = _converter(udt.sqlType())
  
              if conv is None:
  
                  def convert_udt(value: Any) -> Any:
                      return udt.serialize(value)
  
              else:
  
                  def convert_udt(value: Any) -> Any:
                      return conv(udt.serialize(value))  # type: ignore[misc]
  
              return convert_udt
  
          return None
  
      conv = _converter(data_type)
      if conv is not None:
          return lambda pser: pser.apply(  # type: ignore[return-value]
              lambda x: conv(x) if x is not None else None  # type: ignore[misc]
          )
      else:
          return lambda pser: pser
  
imports: import datetime
  import itertools
  from typing import Any, Callable, Iterable, List, Optional, Union, TYPE_CHECKING
  from pyspark.sql.types import cast, BooleanType, ByteType, ShortType, IntegerType, IntegralType, LongType, FloatType, DoubleType, DecimalType, StringType, BinaryType, DateType, TimestampType, TimestampNTZType, DayTimeIntervalType, ArrayType, MapType, StructType, StructField, NullType, DataType, UserDefinedType, _create_row
  from pyspark.errors import PySparkTypeError, UnsupportedOperationException, PySparkValueError
label_desc: Use image node_id 15 for calling a global function with example usage: _create_converter_from_pandas(data_type) and returns: correct_timestamp, None, function, function, _check_series_convert_timestamps_internal, convert_array, convert_map, convert_struct, list, unknown, convert_timestamp, list, value, unknown, value, list, unknown, ts, convert_udt, list, value, unknown, value, unknown, unknown, unknown, dict, unknown, unknown, udt, conv, dict, value, unknown, value