[FLINK-27586][python] Support non-keyed co-broadcast processing #19878

Vancior · 2022-06-06T03:06:27Z

What is the purpose of the change

This PR implements non-keyed co-broadcast processing interfaces, including BroadcastStream, BroadcastConnectedStream and BroadcastProcessFunction.

Brief change log

add BroadcastStream, BroadcastConnectedStream, BroadcastProcessFunction and related internal classes
add an distinct CO_BROADCAST_PROCESS function type for broadcast processing
introduce PythonBroadcastStateTransformation and corresponding translator to translate python broadcast process into different operators for batch or streaming

Verifying this change

This change added tests and can be verified as follows:

test_co_broadcast_process in test_data_stream.py

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (yes)
The serializers: (no)
The runtime per-record code paths (performance sensitive): (no)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (no)
The S3 file system connector: (no)

Documentation

Does this pull request introduce a new feature? (yes)
If yes, how is the feature documented? (Python generated doc)

flinkbot · 2022-06-06T03:10:01Z

CI report:

f98551b Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

HuangXingBo

@Vancior Thanks a lot for the PR. I have left some comments.

flink-python/pyflink/datastream/data_stream.py

HuangXingBo · 2022-06-13T03:28:36Z

flink-python/pyflink/datastream/data_stream.py

+        JPythonBroadcastStateTransformation = (
+            gateway.jvm.org.apache.flink.streaming.api.transformations.python
+        ).PythonBroadcastStateTransformation
+        j_state_names = ListConverter().convert(


Does the typeinfo set in MapDescriptor have any effect?

Actually no, but the base class of PythonBroadcastStateTransformation, AbstractBroadcastStateTransformation needs this for constructor, so this's just for expected behavior of the base class, althrough in PyFlink we are not restricting which broadcast states can be accessed during runtime.

flink-python/pyflink/datastream/data_stream.py

HuangXingBo · 2022-06-13T03:40:16Z

flink-python/pyflink/fn_execution/beam/beam_operations.py

+            serialized_fn.map_state_write_cache_size,
+        )
+    else:
+        operator_state_backend = None


When serialized_fn is not a instance of flink_fn_execution_pb2.UserDefinedDataStreamFunction

Only DataStream API can use operator state, so I guess that's okay

flink-python/pyflink/fn_execution/datastream/operations.py

...rg/apache/flink/streaming/api/transformations/python/PythonBroadcastStateTransformation.java

HuangXingBo · 2022-06-13T06:24:15Z

flink-python/pyflink/datastream/data_stream.py

        raise Exception('Cannot override partitioning for KeyedStream.')

-    def broadcast(self) -> 'DataStream':
+    def broadcast(self, *args) -> Union['DataStream', 'BroadcastStream']:


Are we not going to support broadcast on keyedstream?

Yes, that's will be another PR.

HuangXingBo · 2022-06-17T02:35:31Z

flink-python/pyflink/datastream/data_stream.py

+    def broadcast(self, *args) -> 'BroadcastStream':
+        pass
+
+    def broadcast(self, *args):


add the typehint too?

HuangXingBo · 2022-06-17T02:35:58Z

flink-python/pyflink/datastream/data_stream.py

+        pass
+
+    @overload
+    def broadcast(self, *args) -> 'BroadcastStream':


add the typeint MapStateDescriptor?

HuangXingBo · 2022-06-17T02:36:48Z

flink-python/pyflink/datastream/data_stream.py

        raise Exception('Cannot override partitioning for KeyedStream.')

-    def broadcast(self) -> 'DataStream':
+    def broadcast(self, *args):


HuangXingBo · 2022-06-17T02:38:01Z

flink-python/pyflink/datastream/data_stream.py

+    def connect(self, ds: 'BroadcastStream') -> 'BroadcastConnectedStream':
+        pass
+
+    def connect(self, ds):


add the typeint Union['DataStream', 'BroadcastStream']?

HuangXingBo

@Vancior I have only one comment. And I think you can rebase master and push again.

…Flink

…Flink This closes apache#19878.

flinkbot added the component=API/Python label Jun 6, 2022

HuangXingBo reviewed Jun 13, 2022

View reviewed changes

HuangXingBo reviewed Jun 17, 2022

View reviewed changes

Vancior force-pushed the feat/py_broadcast_process branch from 25d5966 to 4404d2a Compare June 27, 2022 01:43

HuangXingBo reviewed Jun 27, 2022

View reviewed changes

HuangXingBo approved these changes Jun 29, 2022

View reviewed changes

vancior98 added 12 commits June 29, 2022 10:10

[FLINK-27657][python] Implement remote operator state backend in PyFlink

9624489

delete interfaces in OperatorStateStore & slightly refactor StateHandler

f79ac25

[FLINK-27586][python] Support non-keyed co-broadcast processing in Py…

5c0952a

…Flink

resolve conflict & fix test cases after rebasing

9481ae7

revert messy imports after rebasing

86b4204

fix typo

89776ea

update according to review

66b07be

update type hint for doc

401221d

update type hint for doc & upgrade sphinx to 3.5.4

23aecf3

fix Sphinx dependency conflict

3458540

upgrade flake8 to support typing.overload

f37d14a

process_broadcast_element could yield values too

f98551b

Vancior force-pushed the feat/py_broadcast_process branch from 4404d2a to f98551b Compare June 29, 2022 02:17

Vancior changed the title ~~[FLINK-27584][python] Support non-keyed co-broadcast processing~~ [FLINK-27586][python] Support non-keyed co-broadcast processing Jun 30, 2022

HuangXingBo closed this in dfdf7af Jun 30, 2022

zhuyufeng0809 pushed a commit to zhuyufeng0809/flink that referenced this pull request Jul 1, 2022

[FLINK-27586][python] Support non-keyed co-broadcast processing in Py…

8f956a6

…Flink This closes apache#19878.

ericccarlson pushed a commit to ericccarlson/flink that referenced this pull request Jul 11, 2022

[FLINK-27586][python] Support non-keyed co-broadcast processing in Py…

fc5956f

…Flink This closes apache#19878.

liujiawinds pushed a commit to liujiawinds/flink that referenced this pull request Jul 22, 2022

[FLINK-27586][python] Support non-keyed co-broadcast processing in Py…

a4711b0

…Flink This closes apache#19878.

MartijnVisser pushed a commit to sunshineJK/flink that referenced this pull request Jul 29, 2022

[FLINK-27586][python] Support non-keyed co-broadcast processing in Py…

4c01d04

…Flink This closes apache#19878.

[FLINK-27586][python] Support non-keyed co-broadcast processing #19878

[FLINK-27586][python] Support non-keyed co-broadcast processing #19878

Uh oh!

Conversation

Vancior commented Jun 6, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Uh oh!

flinkbot commented Jun 6, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

HuangXingBo left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

HuangXingBo left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

Vancior commented Jun 6, 2022 •

edited

Loading

flinkbot commented Jun 6, 2022 •

edited

Loading