code: def pandas_to_arrow(schema, timezone, field_types, series):
      import pyarrow as pa
      import pandas as pd
  
      def create_array(s, t):
          try:
              return pa.Array.from_pandas(s, mask=s.isnull(), type=t)
          except pa.ArrowException as e:
              error_msg = (
                  "Exception thrown when converting pandas.Series (%s) to "
                  "pyarrow.Array (%s)."
              )
              raise RuntimeError(error_msg % (s.dtype, t), e)
  
      arrays = []
      for i in range(len(schema)):
          s = series[i]
          field_type = field_types[i]
          schema_type = schema.types[i]
          if type(s) == pd.DataFrame:
              array_names = [
                  (
                      create_array(s[s.columns[j]], field.type),
                      field.name,
                  )
                  for j, field in enumerate(schema_type)
              ]
              struct_arrays, struct_names = zip(*array_names)
              arrays.append(
                  pa.StructArray.from_arrays(
                      struct_arrays, struct_names
                  )
              )
          else:
              arrays.append(
                  create_array(
                      tz_convert_to_internal(s, field_type, timezone),
                      schema_type,
                  )
              )
      return pa.RecordBatch.from_arrays(arrays, schema=schema)
  
file_path: flink/flink-python/pyflink/table/utils.py
imports: import ast
  from pyflink.common.types import RowKind
  from pyflink.java_gateway import get_gateway
  from pyflink.table.types import DataType, LocalZonedTimestampType, Row, RowType, TimeType, DateType, ArrayType, MapType, TimestampType, FloatType, RawType
  from pyflink.util.java_utils import to_jarray
  import datetime
  import pickle
label_desc: Use image node_id 1 for calling a global function with example usage: pandas_to_arrow(schema, timezone, field_types, series) and returns: pa, pa