spark-sql-streaming-FlatMapGroupsWithState.adoc

FlatMapGroupsWithState Unary Logical Operator

FlatMapGroupsWithState is a unary logical operator that is created to represent the following operators:

• KeyValueGroupedDataset.mapGroupsWithState

• KeyValueGroupedDataset.flatMapGroupsWithState

Note	A unary logical operator (UnaryNode) is a logical operator with a single child logical operator. Read up on UnaryNode (and logical operators in general) in The Internals of Spark SQL book.

FlatMapGroupsWithState is resolved (translated) to:

• FlatMapGroupsWithStateExec unary physical operator for streaming datasets (in FlatMapGroupsWithStateStrategy execution planning strategy)

• MapGroupsExec physical operator for batch datasets (in BasicOperators execution planning strategy)

Creating SerializeFromObject with FlatMapGroupsWithState — apply Factory Method

apply[K: Encoder, V: Encoder, S: Encoder, U: Encoder](
  func: (Any, Iterator[Any], LogicalGroupState[Any]) => Iterator[Any],
  groupingAttributes: Seq[Attribute],
  dataAttributes: Seq[Attribute],
  outputMode: OutputMode,
  isMapGroupsWithState: Boolean,
  timeout: GroupStateTimeout,
  child: LogicalPlan): LogicalPlan

apply creates a SerializeFromObject logical operator with a FlatMapGroupsWithState as its child logical operator.

Internally, apply creates SerializeFromObject object consumer (aka unary logical operator) with FlatMapGroupsWithState logical plan.

Internally, apply finds ExpressionEncoder for the type S and creates a FlatMapGroupsWithState with UnresolvedDeserializer for the types K and V.

In the end, apply creates a SerializeFromObject object consumer with the FlatMapGroupsWithState.

Note	apply is used in KeyValueGroupedDataset.flatMapGroupsWithState operator.

Creating FlatMapGroupsWithState Instance

FlatMapGroupsWithState takes the following when created:

• State function of type (Any, Iterator[Any], LogicalGroupState[Any]) ⇒ Iterator[Any]

• Key deserializer Catalyst expression

• Value deserializer Catalyst expression

• Grouping attributes

• Data attributes

• Output object attribute

• State ExpressionEncoder

• Output mode

• isMapGroupsWithState flag (default: false)

• GroupStateTimeout

• Child logical operator