[FLINK-22913] Support Python UDF chaining in Python DataStream API

• Type: Improvement
• Status: Closed
• Priority: Major
• Resolution: Done
• Fix Version/s: 1.14.0
• Component/s: API / Python

Release Note

The job graph of Python DataStream API jobs may be different from before as the Python functions will be chained as much as possible to optimize the performance. You could disable Python functions chaining by setting 'python.operator-chaining.enabled' as 'false' explicitly.

• Assignee: Dian Fu
• Reporter: Dian Fu
• Created: 08/Jun/21 06:04
• Updated: 13/Sep/21 12:02
• Resolved: 24/Aug/21 09:29

Description

Currently, for the following job:

ds = ..
ds.map(map_func1)
    .map(map_func2)

The Python function `map_func1` and `map_func2` will runs in separate Python workers and the result of `map_func1` will be transferred to JVM and then transferred to `map_func2` which may resides in another Python worker. This introduces redundant communication and serialization/deserialization overhead.

Sub-Tasks

• Remove ProcessFunctionOperation: Closed
• Support to decode a single record for the Python coder: Closed
• Separate data and timer connection into different channels for Python DataStream API operators: Closed
• Remove Python operators PythonFlatMapOperator/PythonMapOperator/PythonPartitionCustomOperator and use PythonProcessOperator instead: Closed
• Introduce PythonCoProcessOperator and remove PythonCoFlatMapOperator & PythonCoMapOperator: Closed
• Support to chain the Python DataStream operators as much as possible: Closed
• Remove PythonTimestampsAndWatermarksOperator: Closed
• Add documentation about Python DataStream API chaining optimization: Closed
• Testing Python DataStream API chaining functionality: Closed