[SPARK-40431] Introduce "Arbitrary Stateful Processing" in Structured Streaming with Python

• Type: Umbrella
• Status: Open
• Priority: Major
• Resolution: Unresolved
• Affects Version/s: 3.4.0
• Component/s: Structured Streaming
• Assignee: Unassigned
• Reporter: Jungtaek Lim
• Created: 15/Sep/22 02:01
• Updated: 20/Oct/22 01:26

Description

This is a part of effort for SPARK-39590, python API parity in Structured Streaming.

Most of public APIs are available in both Scala/Java Spark and PySpark, but we have a huge gap on streaming workload in PySpark as we don't have matching API for flatMapGroupsWithState in PySpark.

This ticket is to track the effort.

Issue Links

• duplicates: SPARK-39590 Python API Parity in Structure Streaming

Sub-Tasks

• Introduce GroupStateImpl and GroupStateTimeout in PySpark: Resolved
• Add toJVMRow in PythonSQLUtils to convert pickled PySpark Row to JVM Row: Resolved
• Implement applyInPandasWithState in PySpark: Resolved
• Add test suites for applyInPandasWithState in PySpark: Resolved
• Support additionalDuration parameter in GroupState.setTimeoutTimestamp: Open
• Support string representation of durationMs in GroupState.setTimeoutDuration: Open
• Support applyInPandasWithState in batch query: Open
• Support initial state in applyInPandasWithState: Open
• Construct an example of applyInPandasWithState in examples directory: Resolved
• Construct a test case to verify fault-tolerance semantic with random python worker failures: Resolved
• Improving testability of GroupState in applyInPandasWithState: Open