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 through an iterator of (iterator of pandas DataFrame, state), serialize them to Arrow
  RecordBatches, and write batches to stream.
code: def dump_stream(self, iterator, stream):
      """
      Read through an iterator of (iterator of pandas DataFrame, state), serialize them to Arrow
      RecordBatches, and write batches to stream.
      """
  
      import pandas as pd
      import pyarrow as pa
  
      def construct_state_pdf(state):
          """
          Construct a pandas DataFrame from the state instance.
          """
  
          state_properties = state.json().encode("utf-8")
          state_key_row_as_binary = state._keyAsUnsafe
          if state.exists:
              state_object = self.pickleSer.dumps(
                  state._value_schema.toInternal(state._value)
              )
          else:
              state_object = None
          state_old_timeout_timestamp = state.oldTimeoutTimestamp
  
          state_dict = {
              "properties": [
                  state_properties,
              ],
              "keyRowAsUnsafe": [
                  state_key_row_as_binary,
              ],
              "object": [
                  state_object,
              ],
              "oldTimeoutTimestamp": [
                  state_old_timeout_timestamp,
              ],
          }
  
          return pd.DataFrame.from_dict(state_dict)
  
      def construct_record_batch(
          pdfs, pdf_data_cnt, pdf_schema, state_pdfs, state_data_cnt
      ):
          """
          Construct a new Arrow RecordBatch based on output pandas DataFrames and states. Each
          one matches to the single struct field for Arrow schema. We also need an extra one to
          indicate array length for data and state, so the return value of Arrow RecordBatch will
          have schema with three fields, in `count`, `data`, `state` order.
          (Readers are expected to access the field via position rather than the name. We do
          not guarantee the name of the field.)
  
          Note that Arrow RecordBatch requires all columns to have all same number of rows,
          hence this function inserts empty data for count/state/data with less elements to
          compensate.
          """
  
          max_data_cnt = max(1, max(pdf_data_cnt, state_data_cnt))
  
          # We only use the first row in the count column, and fill other rows to be the same
          # value, hoping it is more friendly for compression, in case it is needed.
          count_dict = {
              "dataCount": [pdf_data_cnt] * max_data_cnt,
              "stateCount": [state_data_cnt] * max_data_cnt,
          }
          count_pdf = pd.DataFrame.from_dict(count_dict)
  
          empty_row_cnt_in_data = max_data_cnt - pdf_data_cnt
          empty_row_cnt_in_state = max_data_cnt - state_data_cnt
  
          empty_rows_pdf = pd.DataFrame(
              dict.fromkeys(pa.schema(pdf_schema).names),
              index=[x for x in range(0, empty_row_cnt_in_data)],
          )
          empty_rows_state = pd.DataFrame(
              columns=[
                  "properties",
                  "keyRowAsUnsafe",
                  "object",
                  "oldTimeoutTimestamp",
              ],
              index=[x for x in range(0, empty_row_cnt_in_state)],
          )
  
          pdfs.append(empty_rows_pdf)
          state_pdfs.append(empty_rows_state)
  
          merged_pdf = pd.concat(pdfs, ignore_index=True)
          merged_state_pdf = pd.concat(state_pdfs, ignore_index=True)
  
          return self._create_batch(
              [
                  (count_pdf, self.result_count_pdf_arrow_type),
                  (merged_pdf, pdf_schema),
                  (merged_state_pdf, self.result_state_pdf_arrow_type),
              ]
          )
  
      def serialize_batches():
          """
          Read through an iterator of (iterator of pandas DataFrame, state), and serialize them
          to Arrow RecordBatches.
  
          This function does batching on constructing the Arrow RecordBatch; a batch will be
          serialized to the Arrow RecordBatch when the total number of records exceeds the
          configured threshold.
          """
          # a set of variables for the state of current batch which will be converted to Arrow
          # RecordBatch.
          pdfs = []
          state_pdfs = []
          pdf_data_cnt = 0
          state_data_cnt = 0
  
          return_schema = None
  
          for data in iterator:
              # data represents the result of each call of user function
              packaged_result = data[0]
  
              # There are two results from the call of user function:
              # 1) iterator of pandas DataFrame (output)
              # 2) updated state instance
              pdf_iter = packaged_result[0][0]
              state = packaged_result[0][1]
  
              # This is static and won't change across batches.
              return_schema = packaged_result[1]
  
              for pdf in pdf_iter:
                  # We ignore empty pandas DataFrame.
                  if len(pdf) > 0:
                      pdf_data_cnt += len(pdf)
                      pdfs.append(pdf)
  
                      # If the total number of records in current batch exceeds the configured
                      # threshold, time to construct the Arrow RecordBatch from the batch.
                      if (
                          pdf_data_cnt
                          > self.arrow_max_records_per_batch
                      ):
                          batch = construct_record_batch(
                              pdfs,
                              pdf_data_cnt,
                              return_schema,
                              state_pdfs,
                              state_data_cnt,
                          )
  
                          # Reset the variables to start with new batch for further data.
                          pdfs = []
                          state_pdfs = []
                          pdf_data_cnt = 0
                          state_data_cnt = 0
  
                          yield batch
  
              # This has to be performed 'after' evaluating all elements in iterator, so that
              # the user function has been completed and the state is guaranteed to be updated.
              state_pdf = construct_state_pdf(state)
  
              state_pdfs.append(state_pdf)
              state_data_cnt += 1
  
          # processed all output, but current batch may not be flushed yet.
          if pdf_data_cnt > 0 or state_data_cnt > 0:
              batch = construct_record_batch(
                  pdfs,
                  pdf_data_cnt,
                  return_schema,
                  state_pdfs,
                  state_data_cnt,
              )
  
              yield batch
  
      def init_stream_yield_batches(batches):
          """
          This function helps to ensure the requirement for Pandas UDFs - Pandas UDFs require a
          START_ARROW_STREAM before the Arrow stream is sent.
  
          START_ARROW_STREAM should be sent after creating the first record batch so in case of
          an error, it can be sent back to the JVM before the Arrow stream starts.
          """
          should_write_start_length = True
  
          for batch in batches:
              if should_write_start_length:
                  write_int(SpecialLengths.START_ARROW_STREAM, stream)
                  should_write_start_length = False
  
              yield batch
  
      batches_to_write = init_stream_yield_batches(serialize_batches())
  
      return ArrowStreamSerializer.dump_stream(
          self, batches_to_write, stream
      )
  
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 3 for calling the ApplyInPandasWithStateSerializer obj's underlying member method code with example usage: obj.dump_stream(iterator, stream) and returns: ArrowStreamSerializer, pd, self