class_docstr: Serializer used by Python worker to evaluate UDF for applyInPandasWithState.
  
  Parameters
  ----------
  timezone : str
      A timezone to respect when handling timestamp values
  safecheck : bool
      If True, conversion from Arrow to Pandas checks for overflow/truncation
  assign_cols_by_name : bool
      If True, then Pandas DataFrames will get columns by name
  state_object_schema : StructType
      The type of state object represented as Spark SQL type
  arrow_max_records_per_batch : int
      Limit of the number of records that can be written to a single ArrowRecordBatch in memory.
class_docstr_tok: ["Serializer","used","by","Python","worker","to","evaluate","UDF","for","applyInPandasWithState",".","Parameters","--","--","--","--","--","timezone",":","str","A","timezone","to","respect","when","handling","timestamp","values","safecheck",":","bool","If","True",",","conversion","from","Arrow","to","Pandas","checks","for","overflow\/truncation","assign_cols_by_name",":","bool","If","True",",","then","Pandas","DataFrames","will","get","columns","by","name","state_object_schema",":","StructType","The","type","of","state","object","represented","as","Spark","SQL","type","arrow_max_records_per_batch",":","int","Limit","of","the","number","of","records","that","can","be","written","to","a","single","ArrowRecordBatch","in","memory","."]
docstr: Read ArrowRecordBatches from stream, deserialize them to populate a list of pair
  (data chunk, state), and convert the data into a list of pandas.Series.
  
  Please refer the doc of inner function `gen_data_and_state` for more details how
  this function works in overall.
  
  In addition, this function further groups the return of `gen_data_and_state` by the state
  instance (same semantic as grouping by grouping key) and produces an iterator of data
  chunks for each group, so that the caller can lazily materialize the data chunk.
code: def load_stream(self, stream):
      """
      Read ArrowRecordBatches from stream, deserialize them to populate a list of pair
      (data chunk, state), and convert the data into a list of pandas.Series.
  
      Please refer the doc of inner function `gen_data_and_state` for more details how
      this function works in overall.
  
      In addition, this function further groups the return of `gen_data_and_state` by the state
      instance (same semantic as grouping by grouping key) and produces an iterator of data
      chunks for each group, so that the caller can lazily materialize the data chunk.
      """
  
      import pyarrow as pa
      import json
      from itertools import groupby
      from pyspark.sql.streaming.state import GroupState
  
      def construct_state(state_info_col):
          """
          Construct state instance from the value of state information column.
          """
  
          state_info_col_properties = state_info_col["properties"]
          state_info_col_key_row = state_info_col["keyRowAsUnsafe"]
          state_info_col_object = state_info_col["object"]
  
          state_properties = json.loads(state_info_col_properties)
          if state_info_col_object:
              state_object = self.pickleSer.loads(state_info_col_object)
          else:
              state_object = None
          state_properties["optionalValue"] = state_object
  
          return GroupState(
              keyAsUnsafe=state_info_col_key_row,
              valueSchema=self.state_object_schema,
              **state_properties,
          )
  
      def gen_data_and_state(batches):
          """
          Deserialize ArrowRecordBatches and return a generator of
          `(a list of pandas.Series, state)`.
  
          The logic on deserialization is following:
  
          1. Read the entire data part from Arrow RecordBatch.
          2. Read the entire state information part from Arrow RecordBatch.
          3. Loop through each state information:
             3.A. Extract the data out from entire data via the information of data range.
             3.B. Construct a new state instance if the state information is the first occurrence
                  for the current grouping key.
             3.C. Leverage the existing state instance if it is already available for the current
                  grouping key. (Meaning it's not the first occurrence.)
             3.D. Remove the cache of state instance if the state information denotes the data is
                  the last chunk for current grouping key.
  
          This deserialization logic assumes that Arrow RecordBatches contain the data with the
          ordering that data chunks for same grouping key will appear sequentially.
  
          This function must avoid materializing multiple Arrow RecordBatches into memory at the
          same time. And data chunks from the same grouping key should appear sequentially, to
          further group them based on state instance (same state instance will be produced for
          same grouping key).
          """
  
          state_for_current_group = None
  
          for batch in batches:
              batch_schema = batch.schema
              data_schema = pa.schema(
                  [
                      batch_schema[i]
                      for i in range(0, len(batch_schema) - 1)
                  ]
              )
              state_schema = pa.schema(
                  [
                      batch_schema[-1],
                  ]
              )
  
              batch_columns = batch.columns
              data_columns = batch_columns[0:-1]
              state_column = batch_columns[-1]
  
              data_batch = pa.RecordBatch.from_arrays(
                  data_columns, schema=data_schema
              )
              state_batch = pa.RecordBatch.from_arrays(
                  [
                      state_column,
                  ],
                  schema=state_schema,
              )
  
              state_arrow = pa.Table.from_batches(
                  [state_batch]
              ).itercolumns()
              state_pandas = [
                  self.arrow_to_pandas(c) for c in state_arrow
              ][0]
  
              for state_idx in range(0, len(state_pandas)):
                  state_info_col = state_pandas.iloc[state_idx]
  
                  if not state_info_col:
                      # no more data with grouping key + state
                      break
  
                  data_start_offset = state_info_col["startOffset"]
                  num_data_rows = state_info_col["numRows"]
                  is_last_chunk = state_info_col["isLastChunk"]
  
                  if state_for_current_group:
                      # use the state, we already have state for same group and there should be
                      # some data in same group being processed earlier
                      state = state_for_current_group
                  else:
                      # there is no state being stored for same group, construct one
                      state = construct_state(state_info_col)
  
                  if is_last_chunk:
                      # discard the state being cached for same group
                      state_for_current_group = None
                  elif not state_for_current_group:
                      # there's no cached state but expected to have additional data in same group
                      # cache the current state
                      state_for_current_group = state
  
                  data_batch_for_group = data_batch.slice(
                      data_start_offset, num_data_rows
                  )
                  data_arrow = pa.Table.from_batches(
                      [data_batch_for_group]
                  ).itercolumns()
  
                  data_pandas = [
                      self.arrow_to_pandas(c) for c in data_arrow
                  ]
  
                  # state info
                  yield (
                      data_pandas,
                      state,
                  )
  
      _batches = super(ArrowStreamPandasSerializer, self).load_stream(
          stream
      )
  
      data_state_generator = gen_data_and_state(_batches)
  
      # state will be same object for same grouping key
      for _state, _data in groupby(
          data_state_generator, key=lambda x: x[1]
      ):
          yield (
              _data,
              _state,
          )
  
imports: from pyspark.errors import PySparkRuntimeError, PySparkTypeError, PySparkValueError
  from pyspark.loose_version import LooseVersion
  from pyspark.serializers import Serializer, read_int, write_int, UTF8Deserializer, CPickleSerializer
  from pyspark.sql.pandas.types import from_arrow_type, to_arrow_type, _create_converter_from_pandas, _create_converter_to_pandas
  from pyspark.sql.types import DataType, StringType, StructType, BinaryType, StructField, LongType, IntegerType
label_desc: Use image node_id 2 for calling the ApplyInPandasWithStateSerializer obj's underlying member method code with example usage: obj.load_stream(stream) and returns: GroupState