Type Members

1.  class ApplyInPandasWithStatePythonRunner extends BasePythonRunner[InType, OutType] with PythonArrowInput[InType] with PythonArrowOutput[OutType]

   A variant implementation of ArrowPythonRunner to serve the operation applyInPandasWithState.

   A variant implementation of ArrowPythonRunner to serve the operation applyInPandasWithState.

   Unlike normal ArrowPythonRunner which both input and output (executor <-> python worker) are InternalRow, applyInPandasWithState has side data (state information) in both input and output along with data, which requires different struct on Arrow RecordBatch.

2.  class ApplyInPandasWithStateWriter extends BaseStreamingArrowWriter

   This class abstracts the complexity on constructing Arrow RecordBatches for data and state with bin-packing and chunking.

   This class abstracts the complexity on constructing Arrow RecordBatches for data and state with bin-packing and chunking. The caller only need to call the proper public methods of this class startNewGroup, writeRow, finalizeGroup, finalizeData and this class will write the data and state into Arrow RecordBatches with performing bin-pack and chunk internally.

   This class requires that the parameter root has been initialized with the Arrow schema like below: - data fields - state field

   • nested schema (Refer ApplyInPandasWithStateWriter.STATE_METADATA_SCHEMA)

   Please refer the code comment in the implementation to see how the writes of data and state against Arrow RecordBatch work with consideration of bin-packing and chunking.

3.  class BaseStreamingArrowWriter extends AnyRef

   Base class to handle writing data to Arrow stream to Python workers.

   Base class to handle writing data to Arrow stream to Python workers. When the rows for a group exceed the maximum number of records per batch, we chunk the data into multiple batches.

4.  case class FlatMapGroupsInPandasWithStateExec(functionExpr: Expression, groupingAttributes: Seq[Attribute], outAttributes: Seq[Attribute], stateType: StructType, stateInfo: Option[StatefulOperatorStateInfo], stateFormatVersion: Int, outputMode: OutputMode, timeoutConf: GroupStateTimeout, batchTimestampMs: Option[Long], eventTimeWatermarkForLateEvents: Option[Long], eventTimeWatermarkForEviction: Option[Long], skipEmittingInitialStateKeys: Boolean, child: SparkPlan) extends SparkPlan with UnaryExecNode with FlatMapGroupsWithStateExecBase with Product with Serializable

   Physical operator for executing org.apache.spark.sql.catalyst.plans.logical.FlatMapGroupsInPandasWithState

   Physical operator for executing org.apache.spark.sql.catalyst.plans.logical.FlatMapGroupsInPandasWithState

   functionExpr

   function called on each group

   groupingAttributes

   used to group the data

   outAttributes

   used to define the output rows

   stateType

   used to serialize/deserialize state before calling functionExpr

   stateInfo

   StatefulOperatorStateInfo to identify the state store for a given operator.

   stateFormatVersion

   the version of state format.

   outputMode

   the output mode of functionExpr

   timeoutConf

   used to timeout groups that have not received data in a while

   batchTimestampMs

   processing timestamp of the current batch.

   eventTimeWatermarkForLateEvents

   event time watermark for filtering late events

   eventTimeWatermarkForEviction

   event time watermark for state eviction

   skipEmittingInitialStateKeys

   whether to skip emitting initial state df keys

   child

   logical plan of the underlying data

5.  case class ListStateInfo(listState: ListState[Row], schema: StructType, deserializer: Deserializer[Row], serializer: Serializer[Row]) extends Product with Serializable

   Case class to store the information of a list state.

6.  case class MapStateInfo(mapState: MapState[Row, Row], keySchema: StructType, valueSchema: StructType, keyDeserializer: Deserializer[Row], keySerializer: Serializer[Row], valueDeserializer: Deserializer[Row], valueSerializer: Serializer[Row]) extends Product with Serializable

   Case class to store the information of a map state.

7.  class PythonForeachWriter extends ForeachWriter[UnsafeRow]

   The class proceeds as follows:

   The class proceeds as follows:

   • Rows streamed through a process() call on the org.apache.spark.sql.execution.streaming.QueryExecutionThread are buffered in the UnsafeRowBuffer. - The WriterThread streams the buffered data to the Python worker. - Once the streaming query ends, close() is called which signals the buffer to mark the end of streaming input. The streaming query execution thread waits for the WriterThread to complete and throws any exceptions seen by the WriterThread.
8.  class PythonStreamingSourceRunner extends Logging

   This class is a proxy to invoke methods in Python DataSourceStreamReader from JVM.

   This class is a proxy to invoke methods in Python DataSourceStreamReader from JVM. A runner spawns a python worker process. In the main function, set up communication between JVM and python process through socket and create a DataSourceStreamReader instance. In an infinite loop, the python worker process poll information(function name and parameters) from the socket, invoke the corresponding method of StreamReader and send return value to JVM.

9.  class TransformWithStateInPandasDeserializer extends Logging

   A helper class to deserialize state Arrow batches from the state socket in TransformWithStateInPandas.

10.  case class TransformWithStateInPandasExec(functionExpr: Expression, groupingAttributes: Seq[Attribute], output: Seq[Attribute], outputMode: OutputMode, timeMode: TimeMode, stateInfo: Option[StatefulOperatorStateInfo], batchTimestampMs: Option[Long], eventTimeWatermarkForLateEvents: Option[Long], eventTimeWatermarkForEviction: Option[Long], child: SparkPlan, isStreaming: Boolean = true, hasInitialState: Boolean, initialState: SparkPlan, initialStateGroupingAttrs: Seq[Attribute], initialStateSchema: StructType) extends SparkPlan with BinaryExecNode with StateStoreWriter with WatermarkSupport with TransformWithStateMetadataUtils with Product with Serializable

    Physical operator for executing org.apache.spark.sql.catalyst.plans.logical.TransformWithStateInPandas

    Physical operator for executing org.apache.spark.sql.catalyst.plans.logical.TransformWithStateInPandas

    functionExpr

    function called on each group

    groupingAttributes

    used to group the data

    output

    used to define the output rows

    outputMode

    defines the output mode for the statefulProcessor

    timeMode

    The time mode semantics of the stateful processor for timers and TTL.

    stateInfo

    Used to identify the state store for a given operator.

    batchTimestampMs

    processing timestamp of the current batch.

    eventTimeWatermarkForLateEvents

    event time watermark for filtering late events

    eventTimeWatermarkForEviction

    event time watermark for state eviction

    child

    the physical plan for the underlying data

    isStreaming

    defines whether the query is streaming or batch

    hasInitialState

    defines whether the query has initial state

    initialState

    the physical plan for the input initial state

    initialStateGroupingAttrs

    grouping attributes for initial state

    initialStateSchema

    schema for initial state

11.  abstract class TransformWithStateInPandasPythonBaseRunner[I] extends BasePythonRunner[I, ColumnarBatch] with PythonArrowInput[I] with BasicPythonArrowOutput with TransformWithStateInPandasPythonRunnerUtils with Logging

    Base Python runner implementation for TransformWithStateInPandas.

12.  class TransformWithStateInPandasPythonInitialStateRunner extends TransformWithStateInPandasPythonBaseRunner[GroupedInType] with PythonArrowInput[GroupedInType]

    Python runner with initial state in TransformWithStateInPandas.

    Python runner with initial state in TransformWithStateInPandas. Write input data as one InternalRow(inputRow, initialState) in each row in arrow batch.

13.  class TransformWithStateInPandasPythonPreInitRunner extends StreamingPythonRunner with TransformWithStateInPandasPythonRunnerUtils with Logging

    TransformWithStateInPandas driver side Python runner.

    TransformWithStateInPandas driver side Python runner. Similar as executor side runner, will start a new daemon thread on the Python runner to run state server.

14.  class TransformWithStateInPandasPythonRunner extends TransformWithStateInPandasPythonBaseRunner[InType] with PythonArrowInput[InType]

    Python runner with no initial state in TransformWithStateInPandas.

    Python runner with no initial state in TransformWithStateInPandas. Write input data as one single InternalRow in each row in arrow batch.

15.  trait TransformWithStateInPandasPythonRunnerUtils extends Logging

    TransformWithStateInPandas Python runner utils functions for handling a state server in a new daemon thread.

16.  class TransformWithStateInPandasStateServer extends Runnable with Logging

    This class is used to handle the state requests from the Python side.

    This class is used to handle the state requests from the Python side. It runs on a separate thread spawned by TransformWithStateInPandasStateRunner per task. It opens a dedicated socket to process/transfer state related info which is shut down when task finishes or there's an error on opening the socket. It processes following state requests and return responses to the Python side: - Requests for managing explicit grouping key. - Stateful processor requests. - Requests for managing state variables (e.g. valueState).

17.  case class ValueStateInfo(valueState: ValueState[Row], schema: StructType, deserializer: Deserializer[Row]) extends Product with Serializable

    Case class to store the information of a value state.

18.  class WriterThread extends Thread

    Writes the rows buffered in UnsafeRowBuffer to the Python worker.

    Writes the rows buffered in UnsafeRowBuffer to the Python worker. Any exceptions encountered will be cached to be read later by the parent thread.