code: def _create_converter_to_pandas(
      data_type: DataType,
      nullable: bool = True,
      *,
      timezone: Optional[str] = None,
      struct_in_pandas: Optional[str] = None,
      error_on_duplicated_field_names: bool = True,
      timestamp_utc_localized: bool = True,
      ndarray_as_list: bool = False,
  ) -> Callable[["pd.Series"], "pd.Series"]:
      """
      Create a converter of pandas Series that is created from Spark's Python objects,
      or `pyarrow.Table.to_pandas` method.
  
      Parameters
      ----------
      data_type : :class:`DataType`
          The data type corresponding to the pandas Series to be converted.
      nullable : bool, optional
          Whether the column is nullable or not. (default ``True``)
      timezone : str, optional
          The timezone to convert from. If there is a timestamp type, it's required.
      struct_in_pandas : str, optional
          How to handle struct type. If there is a struct type, it's required.
          When ``row``, :class:`Row` object will be used.
          When ``dict``, :class:`dict` will be used. If there are duplicated field names,
          The fields will be suffixed, like `a_0`, `a_1`.
          Must be one of: ``row``, ``dict``.
      error_on_duplicated_field_names : bool, optional
          Whether raise an exception when there are duplicated field names.
          (default ``True``)
      timestamp_utc_localized : bool, optional
          Whether the timestamp values are localized to UTC or not.
          The timestamp values from Arrow are localized to UTC,
          whereas the ones from `df.collect()` are localized to the local timezone.
      ndarray_as_list : bool, optional
          Whether `np.ndarray` is converted to a list or not (default ``False``).
  
      Returns
      -------
      The converter of `pandas.Series`
      """
      import numpy as np
      import pandas as pd
  
      pandas_type = _to_corrected_pandas_type(data_type)
  
      if pandas_type is not None:
          # SPARK-21766: if an integer field is nullable and has null values, it can be
          # inferred by pandas as a float column. If we convert the column with NaN back
          # to integer type e.g., np.int16, we will hit an exception. So we use the
          # pandas-inferred float type, rather than the corrected type from the schema
          # in this case.
          if isinstance(data_type, IntegralType) and nullable:
  
              def correct_dtype(pser: pd.Series) -> pd.Series:
                  if pser.isnull().any():
                      return pser.astype(np.float64, copy=False)
                  else:
                      return pser.astype(pandas_type, copy=False)
  
          elif isinstance(data_type, BooleanType) and nullable:
  
              def correct_dtype(pser: pd.Series) -> pd.Series:
                  if pser.isnull().any():
                      return pser.astype(object, copy=False)
                  else:
                      return pser.astype(pandas_type, copy=False)
  
          elif isinstance(data_type, TimestampType):
              assert timezone is not None
  
              def correct_dtype(pser: pd.Series) -> pd.Series:
                  if not isinstance(pser.dtype, pd.DatetimeTZDtype):
                      pser = pser.astype(pandas_type, copy=False)
                  return _check_series_convert_timestamps_local_tz(
                      pser, timezone=cast(str, timezone)
                  )
  
          else:
  
              def correct_dtype(pser: pd.Series) -> pd.Series:
                  return pser.astype(pandas_type, copy=False)
  
          return correct_dtype
  
      def _converter(
          dt: DataType,
          _struct_in_pandas: Optional[str],
          _ndarray_as_list: bool,
      ) -> Optional[Callable[[Any], Any]]:
          if isinstance(dt, ArrayType):
              _element_conv = _converter(
                  dt.elementType, _struct_in_pandas, _ndarray_as_list
              )
  
              if _ndarray_as_list:
                  if _element_conv is None:
  
                      def convert_array_ndarray_as_list(
                          value: Any,
                      ) -> Any:
                          # In Arrow Python UDF, ArrayType is converted to `np.ndarray`
                          # whereas a list is expected.
                          return list(value)
  
                  else:
  
                      def convert_array_ndarray_as_list(
                          value: Any,
                      ) -> Any:
                          # In Arrow Python UDF, ArrayType is converted to `np.ndarray`
                          # whereas a list is expected.
                          return [
                              _element_conv(v) if v is not None else None  # type: ignore[misc]
                              for v in value
                          ]
  
                  return convert_array_ndarray_as_list
              else:
                  if _element_conv is None:
                      return None
  
                  def convert_array_ndarray_as_ndarray(
                      value: Any,
                  ) -> Any:
                      if isinstance(value, np.ndarray):
                          # `pyarrow.Table.to_pandas` uses `np.ndarray`.
                          return np.array(
                              [
                                  _element_conv(v) if v is not None else None  # type: ignore[misc]
                                  for v in value
                              ]
                          )
                      else:
                          # otherwise, `list` should be used.
                          return [
                              _element_conv(v) if v is not None else None  # type: ignore[misc]
                              for v in value
                          ]
  
                  return convert_array_ndarray_as_ndarray
  
          elif isinstance(dt, MapType):
              _key_conv = _converter(
                  dt.keyType, _struct_in_pandas, _ndarray_as_list
              )
              _value_conv = _converter(
                  dt.valueType, _struct_in_pandas, _ndarray_as_list
              )
  
              if _key_conv is None and _value_conv is None:
  
                  def convert_map(value: Any) -> Any:
                      # `pyarrow.Table.to_pandas` uses `list` of key-value tuple.
                      # otherwise, `dict` should be used.
                      return dict(value)
  
              else:
  
                  def convert_map(value: Any) -> Any:
                      if isinstance(value, list):
                          # `pyarrow.Table.to_pandas` uses `list` of key-value tuple.
                          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
                          }
                      else:
                          # otherwise, `dict` should be used.
                          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):
              assert _struct_in_pandas is not None
  
              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, _struct_in_pandas, _ndarray_as_list
                  )
                  for f in dt.fields
              ]
  
              if _struct_in_pandas == "row":
                  if all(conv is None for conv in field_convs):
  
                      def convert_struct_as_row(value: Any) -> Any:
                          if isinstance(value, dict):
                              # `pyarrow.Table.to_pandas` uses `dict`.
                              _values = [
                                  value.get(name, None)
                                  for i, name in enumerate(
                                      dedup_field_names
                                  )
                              ]
                              return _create_row(field_names, _values)
                          else:
                              # otherwise, `Row` should be used.
                              return _create_row(field_names, value)
  
                  else:
  
                      def convert_struct_as_row(value: Any) -> Any:
                          if isinstance(value, dict):
                              # `pyarrow.Table.to_pandas` uses `dict`.
                              _values = [
                                  conv(v)
                                  if conv is not None and v is not None
                                  else v
                                  for conv, v in zip(
                                      field_convs,
                                      (
                                          value.get(name, None)
                                          for name in dedup_field_names
                                      ),
                                  )
                              ]
                              return _create_row(field_names, _values)
                          else:
                              # otherwise, `Row` should be used.
                              _values = [
                                  conv(v)
                                  if conv is not None and v is not None
                                  else v
                                  for conv, v in zip(field_convs, value)
                              ]
                              return _create_row(field_names, _values)
  
                  return convert_struct_as_row
  
              elif _struct_in_pandas == "dict":
                  if all(conv is None for conv in field_convs):
  
                      def convert_struct_as_dict(value: Any) -> Any:
                          if isinstance(value, dict):
                              # `pyarrow.Table.to_pandas` uses `dict`.
                              return {
                                  name: value.get(name, None)
                                  for name in dedup_field_names
                              }
                          else:
                              # otherwise, `Row` should be used.
                              return dict(zip(dedup_field_names, value))
  
                  else:
  
                      def convert_struct_as_dict(value: Any) -> Any:
                          if isinstance(value, dict):
                              # `pyarrow.Table.to_pandas` uses `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(name, None)
                                          for name in dedup_field_names
                                      ),
                                  )
                              }
                          else:
                              # otherwise, `Row` should be used.
                              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_as_dict
  
              else:
                  raise PySparkValueError(
                      error_class="UNKNOWN_VALUE_FOR",
                      message_parameters={
                          "var": str(_struct_in_pandas)
                      },
                  )
  
          elif isinstance(dt, TimestampType):
              assert timezone is not None
  
              local_tz: Union[datetime.tzinfo, str] = (
                  datetime.timezone.utc
                  if timestamp_utc_localized
                  else _get_local_timezone()
              )
  
              def convert_timestamp(value: Any) -> Any:
                  if (
                      isinstance(value, datetime.datetime)
                      and value.tzinfo is not None
                  ):
                      ts = pd.Timestamp(value)
                  else:
                      ts = pd.Timestamp(value).tz_localize(local_tz)
                  return ts.tz_convert(timezone).tz_localize(None)
  
              return convert_timestamp
  
          elif isinstance(dt, TimestampNTZType):
  
              def convert_timestamp_ntz(value: Any) -> Any:
                  return pd.Timestamp(value)
  
              return convert_timestamp_ntz
  
          elif isinstance(dt, UserDefinedType):
              udt: UserDefinedType = dt
  
              conv = _converter(
                  udt.sqlType(),
                  _struct_in_pandas="row",
                  _ndarray_as_list=True,
              )
  
              if conv is None:
  
                  def convert_udt(value: Any) -> Any:
                      if hasattr(value, "__UDT__"):
                          assert isinstance(value.__UDT__, type(udt))
                          return value
                      else:
                          return udt.deserialize(value)
  
              else:
  
                  def convert_udt(value: Any) -> Any:
                      if hasattr(value, "__UDT__"):
                          assert isinstance(value.__UDT__, type(udt))
                          return value
                      else:
                          return udt.deserialize(conv(value))  # type: ignore[misc]
  
              return convert_udt
  
          else:
              return None
  
      conv = _converter(data_type, struct_in_pandas, ndarray_as_list)
      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 14 for calling a global function with example usage: _create_converter_to_pandas(data_type, nullable) and returns: correct_dtype, function, function, convert_array_ndarray_as_list, convert_array_ndarray_as_ndarray, convert_map, pser, pser, None, pser, pser, _check_series_convert_timestamps_local_tz, pser, list, unknown, np, unknown, dict, convert_struct_as_row, convert_timestamp, unknown, unknown, convert_struct_as_dict, ts, convert_timestamp_ntz, pd, convert_udt, None, _create_row, _create_row, _create_row, _create_row, unknown, dict, unknown, unknown, value, udt, value, udt