You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by di...@apache.org on 2022/04/26 01:15:08 UTC
[flink] branch master updated (98568a248c0 -> 7eaa2741cc3)
This is an automated email from the ASF dual-hosted git repository.
dianfu pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from 98568a248c0 [hotfix] Remove unused classes
new efa0618e936 [FLINK-26477][python] Support side output in Python DataStream API
new 7eaa2741cc3 [hotfix][python][tests] Remove useless test utility methods
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../docs/dev/datastream/operators/windows.md | 26 +++
docs/content.zh/docs/dev/datastream/side_output.md | 33 ++++
.../docs/dev/datastream/operators/windows.md | 26 +++
docs/content/docs/dev/datastream/side_output.md | 33 ++++
flink-python/pyflink/common/constants.py | 6 +
flink-python/pyflink/datastream/__init__.py | 6 +-
flink-python/pyflink/datastream/data_stream.py | 41 ++++-
flink-python/pyflink/datastream/output_tag.py | 61 ++++++
.../pyflink/datastream/tests/test_data_stream.py | 186 ++++++++++++++++++-
.../pyflink/datastream/tests/test_window.py | 40 +++-
flink-python/pyflink/datastream/window.py | 5 +-
.../fn_execution/beam/beam_operations_fast.pxd | 4 +-
.../fn_execution/beam/beam_operations_fast.pyx | 80 +++++---
.../fn_execution/beam/beam_operations_slow.py | 89 +++++++--
.../fn_execution/datastream/input_handler.py | 60 ++++--
.../pyflink/fn_execution/datastream/operations.py | 129 ++++++++-----
.../datastream/window/window_operator.py | 12 +-
.../pyflink/fn_execution/flink_fn_execution_pb2.py | 105 ++++++-----
.../pyflink/proto/flink-fn-execution.proto | 7 +-
flink-python/pyflink/testing/test_case_utils.py | 16 --
.../java/org/apache/flink/python/Constants.java | 5 +-
.../apache/flink/python/PythonFunctionRunner.java | 6 +-
.../chain/PythonOperatorChainingOptimizer.java | 15 +-
.../apache/flink/python/util/PythonConfigUtil.java | 51 +++++
.../AbstractDataStreamPythonFunctionOperator.java | 69 +++++++
.../AbstractExternalPythonFunctionOperator.java | 12 +-
.../AbstractOneInputPythonFunctionOperator.java | 19 +-
.../AbstractTwoInputPythonFunctionOperator.java | 19 +-
.../operators/python/PythonCoProcessOperator.java | 4 +-
.../python/PythonKeyedCoProcessOperator.java | 4 +-
.../python/PythonKeyedProcessOperator.java | 4 +-
.../operators/python/PythonProcessOperator.java | 4 +-
.../python/collector/RunnerOutputCollector.java | 19 ++
.../beam/BeamDataStreamPythonFunctionRunner.java | 205 ++++++++++++++-------
.../python/beam/BeamPythonFunctionRunner.java | 92 ++++++---
.../flink/streaming/api/utils/ProtoUtils.java | 7 +
...AbstractPythonStreamGroupAggregateOperator.java | 8 +-
.../PythonStreamGroupWindowAggregateOperator.java | 8 +-
...hArrowPythonGroupAggregateFunctionOperator.java | 8 +-
...PythonGroupWindowAggregateFunctionOperator.java | 7 +-
...wPythonOverWindowAggregateFunctionOperator.java | 8 +-
...tractStreamArrowPythonBoundedRangeOperator.java | 8 +-
...stractStreamArrowPythonBoundedRowsOperator.java | 8 +-
...PythonGroupWindowAggregateFunctionOperator.java | 7 +-
.../scalar/PythonScalarFunctionOperator.java | 8 +-
.../arrow/ArrowPythonScalarFunctionOperator.java | 8 +-
.../python/table/PythonTableFunctionOperator.java | 8 +-
.../python/beam/BeamTablePythonFunctionRunner.java | 3 +-
.../python/util/DataStreamTestCollectSink.java | 33 +++-
.../eventtime/PerElementWatermarkGenerator.java | 25 ++-
...ghPythonStreamGroupWindowAggregateOperator.java | 12 +-
.../PassThroughPythonAggregateFunctionRunner.java | 8 +-
.../PassThroughPythonScalarFunctionRunner.java | 8 +-
.../PassThroughPythonTableFunctionRunner.java | 8 +-
...ThroughStreamAggregatePythonFunctionRunner.java | 8 +-
...ghStreamTableAggregatePythonFunctionRunner.java | 8 +-
56 files changed, 1320 insertions(+), 379 deletions(-)
create mode 100644 flink-python/pyflink/datastream/output_tag.py
copy flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/source/OnEventTestWatermarkGenerator.java => flink-python/src/test/java/org/apache/flink/streaming/api/functions/python/eventtime/PerElementWatermarkGenerator.java (55%)
[flink] 02/02: [hotfix][python][tests] Remove useless test utility methods
Posted by di...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dianfu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 7eaa2741cc3fdf81b72740d4abdc1db74b641c47
Author: Dian Fu <di...@apache.org>
AuthorDate: Sun Apr 24 13:45:18 2022 +0800
[hotfix][python][tests] Remove useless test utility methods
---
flink-python/pyflink/testing/test_case_utils.py | 16 ----------------
1 file changed, 16 deletions(-)
diff --git a/flink-python/pyflink/testing/test_case_utils.py b/flink-python/pyflink/testing/test_case_utils.py
index 8abe31f8e4c..a8cbc2a292a 100644
--- a/flink-python/pyflink/testing/test_case_utils.py
+++ b/flink-python/pyflink/testing/test_case_utils.py
@@ -29,7 +29,6 @@ from py4j.java_gateway import JavaObject
from pyflink.common import JobExecutionResult
from pyflink.datastream.execution_mode import RuntimeExecutionMode
-from pyflink.table.sources import CsvTableSource
from pyflink.datastream.stream_execution_environment import StreamExecutionEnvironment
from pyflink.find_flink_home import _find_flink_home, _find_flink_source_root
from pyflink.table.table_environment import TableEnvironment
@@ -111,21 +110,6 @@ class PyFlinkTestCase(unittest.TestCase):
py_list.append(actual.get(i))
return py_list
- @classmethod
- def prepare_csv_source(cls, path, data, data_types, fields):
- if os.path.isfile(path):
- os.remove(path)
- csv_data = ""
- for item in data:
- if isinstance(item, list) or isinstance(item, tuple):
- csv_data += ",".join([str(element) for element in item]) + "\n"
- else:
- csv_data += str(item) + "\n"
- with open(path, 'w') as f:
- f.write(csv_data)
- f.close()
- return CsvTableSource(path, fields, data_types)
-
class PyFlinkStreamTableTestCase(PyFlinkTestCase):
"""
[flink] 01/02: [FLINK-26477][python] Support side output in Python DataStream API
Posted by di...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dianfu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit efa0618e93626a50c69ff0d89c8159bd50c9ee07
Author: Juntao Hu <ma...@gmail.com>
AuthorDate: Wed Apr 6 20:10:58 2022 +0800
[FLINK-26477][python] Support side output in Python DataStream API
This closes #19453.
---
.../docs/dev/datastream/operators/windows.md | 26 +++
docs/content.zh/docs/dev/datastream/side_output.md | 33 ++++
.../docs/dev/datastream/operators/windows.md | 26 +++
docs/content/docs/dev/datastream/side_output.md | 33 ++++
flink-python/pyflink/common/constants.py | 6 +
flink-python/pyflink/datastream/__init__.py | 6 +-
flink-python/pyflink/datastream/data_stream.py | 41 ++++-
flink-python/pyflink/datastream/output_tag.py | 61 ++++++
.../pyflink/datastream/tests/test_data_stream.py | 186 ++++++++++++++++++-
.../pyflink/datastream/tests/test_window.py | 40 +++-
flink-python/pyflink/datastream/window.py | 5 +-
.../fn_execution/beam/beam_operations_fast.pxd | 4 +-
.../fn_execution/beam/beam_operations_fast.pyx | 80 +++++---
.../fn_execution/beam/beam_operations_slow.py | 89 +++++++--
.../fn_execution/datastream/input_handler.py | 60 ++++--
.../pyflink/fn_execution/datastream/operations.py | 129 ++++++++-----
.../datastream/window/window_operator.py | 12 +-
.../pyflink/fn_execution/flink_fn_execution_pb2.py | 105 ++++++-----
.../pyflink/proto/flink-fn-execution.proto | 7 +-
.../java/org/apache/flink/python/Constants.java | 5 +-
.../apache/flink/python/PythonFunctionRunner.java | 6 +-
.../chain/PythonOperatorChainingOptimizer.java | 15 +-
.../apache/flink/python/util/PythonConfigUtil.java | 51 +++++
.../AbstractDataStreamPythonFunctionOperator.java | 69 +++++++
.../AbstractExternalPythonFunctionOperator.java | 12 +-
.../AbstractOneInputPythonFunctionOperator.java | 19 +-
.../AbstractTwoInputPythonFunctionOperator.java | 19 +-
.../operators/python/PythonCoProcessOperator.java | 4 +-
.../python/PythonKeyedCoProcessOperator.java | 4 +-
.../python/PythonKeyedProcessOperator.java | 4 +-
.../operators/python/PythonProcessOperator.java | 4 +-
.../python/collector/RunnerOutputCollector.java | 19 ++
.../beam/BeamDataStreamPythonFunctionRunner.java | 205 ++++++++++++++-------
.../python/beam/BeamPythonFunctionRunner.java | 92 ++++++---
.../flink/streaming/api/utils/ProtoUtils.java | 7 +
...AbstractPythonStreamGroupAggregateOperator.java | 8 +-
.../PythonStreamGroupWindowAggregateOperator.java | 8 +-
...hArrowPythonGroupAggregateFunctionOperator.java | 8 +-
...PythonGroupWindowAggregateFunctionOperator.java | 7 +-
...wPythonOverWindowAggregateFunctionOperator.java | 8 +-
...tractStreamArrowPythonBoundedRangeOperator.java | 8 +-
...stractStreamArrowPythonBoundedRowsOperator.java | 8 +-
...PythonGroupWindowAggregateFunctionOperator.java | 7 +-
.../scalar/PythonScalarFunctionOperator.java | 8 +-
.../arrow/ArrowPythonScalarFunctionOperator.java | 8 +-
.../python/table/PythonTableFunctionOperator.java | 8 +-
.../python/beam/BeamTablePythonFunctionRunner.java | 3 +-
.../python/util/DataStreamTestCollectSink.java | 33 +++-
.../eventtime/PerElementWatermarkGenerator.java | 51 +++++
...ghPythonStreamGroupWindowAggregateOperator.java | 12 +-
.../PassThroughPythonAggregateFunctionRunner.java | 8 +-
.../PassThroughPythonScalarFunctionRunner.java | 8 +-
.../PassThroughPythonTableFunctionRunner.java | 8 +-
...ThroughStreamAggregatePythonFunctionRunner.java | 8 +-
...ghStreamTableAggregatePythonFunctionRunner.java | 8 +-
55 files changed, 1350 insertions(+), 359 deletions(-)
diff --git a/docs/content.zh/docs/dev/datastream/operators/windows.md b/docs/content.zh/docs/dev/datastream/operators/windows.md
index 9441dc82876..0f22b6d0e27 100644
--- a/docs/content.zh/docs/dev/datastream/operators/windows.md
+++ b/docs/content.zh/docs/dev/datastream/operators/windows.md
@@ -1449,6 +1449,16 @@ input
.<windowed transformation>(<window function>)
```
{{< /tab >}}
+{{< tab "Python" >}}
+```python
+input = ... # type: DataStream
+input \
+ .key_by(<key selector>) \
+ .window(<window assigner>) \
+ .allowed_lateness(<time>) \
+ .<windowed transformation>(<window function>)
+```
+{{< /tab >}}
{{< /tabs >}}
{{< hint info >}}
@@ -1495,6 +1505,22 @@ val result = input
val lateStream = result.getSideOutput(lateOutputTag)
```
{{< /tab >}}
+{{< tab "Python" >}}
+```python
+late_output_tag = OutputTag("late-data", type_info)
+
+input = ... # type: DataStream
+
+result = input \
+ .key_by(<key selector>) \
+ .window(<window assigner>) \
+ .allowed_lateness(<time>) \
+ .side_output_late_data(late_output_tag) \
+ .<windowed transformation>(<window function>)
+
+late_stream = result.get_side_output(late_output_tag)
+```
+{{< /tab >}}
{{< /tabs >}}
### 迟到数据的一些考虑
diff --git a/docs/content.zh/docs/dev/datastream/side_output.md b/docs/content.zh/docs/dev/datastream/side_output.md
index 9e5cb35b02e..8b6f4a10de0 100644
--- a/docs/content.zh/docs/dev/datastream/side_output.md
+++ b/docs/content.zh/docs/dev/datastream/side_output.md
@@ -46,6 +46,11 @@ OutputTag<String> outputTag = new OutputTag<String>("side-output") {};
val outputTag = OutputTag[String]("side-output")
```
{{< /tab >}}
+{{< tab "Python" >}}
+```python
+output_tag = OutputTag("side-output", Types.STRING())
+```
+{{< /tab >}}
{{< /tabs >}}
注意 `OutputTag` 是如何根据旁路输出流所包含的元素类型进行类型化的。
@@ -108,6 +113,25 @@ val mainDataStream = input
})
```
{{< /tab >}}
+{{< tab "Python" >}}
+```python
+input = ... # type: DataStream
+output_tag = OutputTag("side-output", Types.STRING())
+
+class MyProcessFunction(ProcessFunction):
+
+ def process_element(self, value: int, ctx: ProcessFunction.Context):
+ # emit data to regular output
+ yield value
+
+ # emit data to side output
+ yield output_tag, "sideout-" + str(value)
+
+
+main_data_stream = input \
+ .process(MyProcessFunction(), Types.INT())
+```
+{{< /tab >}}
{{< /tabs >}}
你可以在 `DataStream` 运算结果上使用 `getSideOutput(OutputTag)` 方法获取旁路输出流。这将产生一个与旁路输出流结果类型一致的 `DataStream`：
@@ -133,6 +157,15 @@ val mainDataStream = ...
val sideOutputStream: DataStream[String] = mainDataStream.getSideOutput(outputTag)
```
{{< /tab >}}
+{{< tab "Python" >}}
+```python
+output_tag = OutputTag("side-output", Types.STRING())
+
+main_data_stream = ... # type: DataStream
+
+side_output_stream = main_data_stream.get_side_output(output_tag) # type: DataStream
+```
+{{< /tab >}}
{{< /tabs >}}
{{< top >}}
diff --git a/docs/content/docs/dev/datastream/operators/windows.md b/docs/content/docs/dev/datastream/operators/windows.md
index 12c723eab4d..19cdc4426b5 100644
--- a/docs/content/docs/dev/datastream/operators/windows.md
+++ b/docs/content/docs/dev/datastream/operators/windows.md
@@ -1494,6 +1494,16 @@ input
.<windowed transformation>(<window function>)
```
{{< /tab >}}
+{{< tab "Python" >}}
+```python
+input = ... # type: DataStream
+input \
+ .key_by(<key selector>) \
+ .window(<window assigner>) \
+ .allowed_lateness(<time>) \
+ .<windowed transformation>(<window function>)
+```
+{{< /tab >}}
{{< /tabs >}}
{{< hint info >}}
@@ -1542,6 +1552,22 @@ val result = input
val lateStream = result.getSideOutput(lateOutputTag)
```
{{< /tab >}}
+{{< tab "Python" >}}
+```python
+late_output_tag = OutputTag("late-data", type_info)
+
+input = ... # type: DataStream
+
+result = input \
+ .key_by(<key selector>) \
+ .window(<window assigner>) \
+ .allowed_lateness(<time>) \
+ .side_output_late_data(late_output_tag) \
+ .<windowed transformation>(<window function>)
+
+late_stream = result.get_side_output(late_output_tag)
+```
+{{< /tab >}}
{{< /tabs >}}
### Late elements considerations
diff --git a/docs/content/docs/dev/datastream/side_output.md b/docs/content/docs/dev/datastream/side_output.md
index fe7dc0d4f78..0f7f9d17719 100644
--- a/docs/content/docs/dev/datastream/side_output.md
+++ b/docs/content/docs/dev/datastream/side_output.md
@@ -50,6 +50,11 @@ OutputTag<String> outputTag = new OutputTag<String>("side-output") {};
val outputTag = OutputTag[String]("side-output")
```
{{< /tab >}}
+{{< tab "Python" >}}
+```python
+output_tag = OutputTag("side-output", Types.STRING())
+```
+{{< /tab >}}
{{< /tabs >}}
Notice how the `OutputTag` is typed according to the type of elements that the side output stream
@@ -115,6 +120,25 @@ val mainDataStream = input
})
```
{{< /tab >}}
+{{< tab "Python" >}}
+```python
+input = ... # type: DataStream
+output_tag = OutputTag("side-output", Types.STRING())
+
+class MyProcessFunction(ProcessFunction):
+
+ def process_element(self, value: int, ctx: ProcessFunction.Context):
+ # emit data to regular output
+ yield value
+
+ # emit data to side output
+ yield output_tag, "sideout-" + str(value)
+
+
+main_data_stream = input \
+ .process(MyProcessFunction(), Types.INT())
+```
+{{< /tab >}}
{{< /tabs >}}
For retrieving the side output stream you use `getSideOutput(OutputTag)`
@@ -142,6 +166,15 @@ val mainDataStream = ...
val sideOutputStream: DataStream[String] = mainDataStream.getSideOutput(outputTag)
```
{{< /tab >}}
+{{< tab "Python" >}}
+```python
+output_tag = OutputTag("side-output", Types.STRING())
+
+main_data_stream = ... # type: DataStream
+
+side_output_stream = main_data_stream.get_side_output(output_tag) # type: DataStream
+```
+{{< /tab >}}
{{< /tabs >}}
{{< top >}}
diff --git a/flink-python/pyflink/common/constants.py b/flink-python/pyflink/common/constants.py
index 93f6407556e..25f4b895ed8 100644
--- a/flink-python/pyflink/common/constants.py
+++ b/flink-python/pyflink/common/constants.py
@@ -19,7 +19,13 @@
A constant holding the maximum value a long can have, 2^63 – 1.
"""
MAX_LONG_VALUE = 0x7fffffffffffffff
+
"""
A constant holding the minimum value a long can have, -2^63
"""
MIN_LONG_VALUE = - MAX_LONG_VALUE - 1
+
+"""
+Output tag for main output.
+"""
+DEFAULT_OUTPUT_TAG = ""
diff --git a/flink-python/pyflink/datastream/__init__.py b/flink-python/pyflink/datastream/__init__.py
index dcc77d8b784..5beeacd74eb 100644
--- a/flink-python/pyflink/datastream/__init__.py
+++ b/flink-python/pyflink/datastream/__init__.py
@@ -183,6 +183,8 @@ Other important classes:
Interface for implementing user defined sink functionality.
- :class:`SourceFunction`:
Interface for implementing user defined source functionality.
+ - :class:`OutputTag`:
+ Tag with a name and type for identifying side output of an operator
"""
from pyflink.datastream.checkpoint_config import CheckpointConfig, ExternalizedCheckpointCleanup
from pyflink.datastream.checkpointing_mode import CheckpointingMode
@@ -210,6 +212,7 @@ from pyflink.datastream.functions import ProcessFunction
from pyflink.datastream.timerservice import TimerService
from pyflink.datastream.window import Window, TimeWindow, CountWindow, WindowAssigner, \
MergingWindowAssigner, TriggerResult, Trigger
+from pyflink.datastream.output_tag import OutputTag
__all__ = [
'StreamExecutionEnvironment',
@@ -263,5 +266,6 @@ __all__ = [
'SourceFunction',
'SinkFunction',
'SlotSharingGroup',
- 'MemorySize'
+ 'MemorySize',
+ 'OutputTag'
]
diff --git a/flink-python/pyflink/datastream/data_stream.py b/flink-python/pyflink/datastream/data_stream.py
index 8d3e48dc625..73a0e169998 100644
--- a/flink-python/pyflink/datastream/data_stream.py
+++ b/flink-python/pyflink/datastream/data_stream.py
@@ -18,7 +18,7 @@
import typing
import uuid
from enum import Enum
-from typing import Callable, Union, List, cast
+from typing import Callable, Union, List, cast, Optional
from pyflink.common import typeinfo, ExecutionConfig, Row
from pyflink.common.typeinfo import RowTypeInfo, Types, TypeInformation, _from_java_type
@@ -36,6 +36,7 @@ from pyflink.datastream.functions import (_get_python_env, FlatMapFunction, MapF
InternalSingleValueWindowFunction,
InternalSingleValueProcessWindowFunction,
PassThroughWindowFunction, AggregateFunction)
+from pyflink.datastream.output_tag import OutputTag
from pyflink.datastream.slot_sharing_group import SlotSharingGroup
from pyflink.datastream.state import ValueStateDescriptor, ValueState, ListStateDescriptor, \
StateDescriptor, ReducingStateDescriptor, AggregatingStateDescriptor
@@ -789,6 +790,18 @@ class DataStream(object):
j_data_stream_sink = self._align_output_type()._j_data_stream.print()
return DataStreamSink(j_data_stream_sink)
+ def get_side_output(self, output_tag: OutputTag) -> 'DataStream':
+ """
+ Gets the :class:`DataStream` that contains the elements that are emitted from an operation
+ into the side output with the given :class:`OutputTag`.
+
+ :param output_tag: output tag for the side stream
+ :return: The DataStream with specified output tag
+
+ .. versionadded:: 1.16.0
+ """
+ return DataStream(self._j_data_stream.getSideOutput(output_tag.get_java_output_tag()))
+
def _apply_chaining_optimization(self):
"""
Chain the Python operators if possible.
@@ -1624,6 +1637,7 @@ class WindowedStream(object):
self._keyed_stream = keyed_stream
self._window_assigner = window_assigner
self._allowed_lateness = 0
+ self._late_data_output_tag = None # type: Optional[OutputTag]
self._window_trigger = None # type: Trigger
def get_execution_environment(self):
@@ -1650,6 +1664,30 @@ class WindowedStream(object):
self._allowed_lateness = time_ms
return self
+ def side_output_late_data(self, output_tag: OutputTag):
+ """
+ Send late arriving data to the side output identified by the given :class:`OutputTag`. Data
+ is considered late after the watermark has passed the end of the window plus the allowed
+ lateness set using :func:`allowed_lateness`.
+
+ You can get the stream of late data using :func:`~DataStream.get_side_output` on the
+ :class:`DataStream` resulting from the windowed operation with the same :class:`OutputTag`.
+
+ Example:
+ ::
+
+ >>> tag = OutputTag("late-data", Types.TUPLE([Types.INT(), Types.STRING()]))
+ >>> main_stream = ds.key_by(lambda x: x[1]) \\
+ ... .window(TumblingEventTimeWindows.of(Time.seconds(5))) \\
+ ... .side_output_late_data(tag) \\
+ ... .reduce(lambda a, b: a[0] + b[0], b[1])
+ >>> late_stream = main_stream.get_side_output(tag)
+
+ .. versionadded:: 1.16.0
+ """
+ self._late_data_output_tag = output_tag
+ return self
+
def reduce(self,
reduce_function: Union[Callable, ReduceFunction],
window_function: Union[WindowFunction, ProcessWindowFunction] = None,
@@ -1823,6 +1861,7 @@ class WindowedStream(object):
self._window_assigner,
self._window_trigger,
self._allowed_lateness,
+ self._late_data_output_tag,
window_state_descriptor,
window_serializer,
internal_window_function)
diff --git a/flink-python/pyflink/datastream/output_tag.py b/flink-python/pyflink/datastream/output_tag.py
new file mode 100644
index 00000000000..78da82d4b4d
--- /dev/null
+++ b/flink-python/pyflink/datastream/output_tag.py
@@ -0,0 +1,61 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+from typing import Optional, Union
+
+from pyflink.common.typeinfo import TypeInformation, Types, RowTypeInfo
+from pyflink.java_gateway import get_gateway
+
+
+class OutputTag(object):
+ """
+ An :class:`OutputTag` is a typed and named tag to use for tagging side outputs of an operator.
+
+ Example:
+ ::
+
+ # Explicitly specify output type
+ >>> info = OutputTag("late-data", Types.TUPLE([Types.STRING(), Types.LONG()]))
+ # Implicitly wrap list to Types.ROW
+ >>> info_row = OutputTag("row", [Types.STRING(), Types.LONG()])
+ # Implicitly use pickle serialization
+ >>> info_side = OutputTag("side")
+ # ERROR: tag id cannot be empty string (extra requirement for Python API)
+ >>> info_error = OutputTag("")
+
+ """
+
+ def __init__(self, tag_id: str, type_info: Optional[Union[TypeInformation, list]] = None):
+ if not tag_id:
+ raise ValueError("OutputTag tag_id cannot be None or empty string")
+ self.tag_id = tag_id
+ if type_info is None:
+ self.type_info = Types.PICKLED_BYTE_ARRAY()
+ elif isinstance(type_info, list):
+ self.type_info = RowTypeInfo(type_info)
+ elif not isinstance(type_info, TypeInformation):
+ raise TypeError("OutputTag type_info must be None, list or TypeInformation")
+ else:
+ self.type_info = type_info
+
+ def get_java_output_tag(self):
+ gateway = get_gateway()
+ j_obj = gateway.jvm.org.apache.flink.util.OutputTag(self.tag_id,
+ self.type_info.get_java_type_info())
+ # deal with serializability
+ self.type_info._j_typeinfo = None
+ return j_obj
diff --git a/flink-python/pyflink/datastream/tests/test_data_stream.py b/flink-python/pyflink/datastream/tests/test_data_stream.py
index 4756b22f6f9..05e25008d51 100644
--- a/flink-python/pyflink/datastream/tests/test_data_stream.py
+++ b/flink-python/pyflink/datastream/tests/test_data_stream.py
@@ -29,7 +29,8 @@ from pyflink.datastream.data_stream import DataStream
from pyflink.datastream.functions import (AggregateFunction, CoMapFunction, CoFlatMapFunction,
MapFunction, FilterFunction, FlatMapFunction,
KeyedCoProcessFunction, KeyedProcessFunction, KeySelector,
- ProcessFunction, ReduceFunction)
+ ProcessFunction, ReduceFunction, CoProcessFunction)
+from pyflink.datastream.output_tag import OutputTag
from pyflink.datastream.state import (ValueStateDescriptor, ListStateDescriptor, MapStateDescriptor,
ReducingStateDescriptor, ReducingState, AggregatingState,
AggregatingStateDescriptor, StateTtlConfig)
@@ -755,6 +756,189 @@ class DataStreamTests(object):
expected = ['(1,hi)', '(2,hello)', '(4,hi)', '(6,hello)', '(9,hi)', '(12,hello)']
self.assert_equals_sorted(expected, results)
+ def test_process_side_output(self):
+ tag = OutputTag("side", Types.INT())
+
+ ds = self.env.from_collection([('a', 0), ('b', 1), ('c', 2)],
+ type_info=Types.ROW([Types.STRING(), Types.INT()]))
+
+ class MyProcessFunction(ProcessFunction):
+
+ def process_element(self, value, ctx: 'ProcessFunction.Context'):
+ yield value[0]
+ yield tag, value[1]
+
+ ds2 = ds.process(MyProcessFunction(), output_type=Types.STRING())
+ main_sink = DataStreamTestSinkFunction()
+ ds2.add_sink(main_sink)
+ side_sink = DataStreamTestSinkFunction()
+ ds2.get_side_output(tag).add_sink(side_sink)
+
+ self.env.execute("test_process_side_output")
+ main_expected = ['a', 'b', 'c']
+ self.assert_equals_sorted(main_expected, main_sink.get_results())
+ side_expected = ['0', '1', '2']
+ self.assert_equals_sorted(side_expected, side_sink.get_results())
+
+ def test_side_output_chained_with_upstream_operator(self):
+ tag = OutputTag("side", Types.INT())
+
+ ds = self.env.from_collection([('a', 0), ('b', 1), ('c', 2)],
+ type_info=Types.ROW([Types.STRING(), Types.INT()]))
+
+ class MyProcessFunction(ProcessFunction):
+
+ def process_element(self, value, ctx: 'ProcessFunction.Context'):
+ yield value[0]
+ yield tag, value[1]
+
+ ds2 = ds.map(lambda e: (e[0], e[1]+1)) \
+ .process(MyProcessFunction(), output_type=Types.STRING())
+ main_sink = DataStreamTestSinkFunction()
+ ds2.add_sink(main_sink)
+ side_sink = DataStreamTestSinkFunction()
+ ds2.get_side_output(tag).add_sink(side_sink)
+
+ self.env.execute("test_side_output_chained_with_upstream_operator")
+ main_expected = ['a', 'b', 'c']
+ self.assert_equals_sorted(main_expected, main_sink.get_results())
+ side_expected = ['1', '2', '3']
+ self.assert_equals_sorted(side_expected, side_sink.get_results())
+
+ def test_process_multiple_side_output(self):
+ tag1 = OutputTag("side1", Types.INT())
+ tag2 = OutputTag("side2", Types.STRING())
+
+ ds = self.env.from_collection([('a', 0), ('b', 1), ('c', 2)],
+ type_info=Types.ROW([Types.STRING(), Types.INT()]))
+
+ class MyProcessFunction(ProcessFunction):
+
+ def process_element(self, value, ctx: 'ProcessFunction.Context'):
+ yield value[0]
+ yield tag1, value[1]
+ yield tag2, value[0] + str(value[1])
+
+ ds2 = ds.process(MyProcessFunction(), output_type=Types.STRING())
+ main_sink = DataStreamTestSinkFunction()
+ ds2.add_sink(main_sink)
+ side1_sink = DataStreamTestSinkFunction()
+ ds2.get_side_output(tag1).add_sink(side1_sink)
+ side2_sink = DataStreamTestSinkFunction()
+ ds2.get_side_output(tag2).add_sink(side2_sink)
+
+ self.env.execute("test_process_multiple_side_output")
+ main_expected = ['a', 'b', 'c']
+ self.assert_equals_sorted(main_expected, main_sink.get_results())
+ side1_expected = ['0', '1', '2']
+ self.assert_equals_sorted(side1_expected, side1_sink.get_results())
+ side2_expected = ['a0', 'b1', 'c2']
+ self.assert_equals_sorted(side2_expected, side2_sink.get_results())
+
+ def test_co_process_side_output(self):
+ tag = OutputTag("side", Types.INT())
+
+ class MyCoProcessFunction(CoProcessFunction):
+
+ def process_element1(self, value, ctx: 'CoProcessFunction.Context'):
+ yield value[0]
+ yield tag, value[1]
+
+ def process_element2(self, value, ctx: 'CoProcessFunction.Context'):
+ yield value[1]
+ yield tag, value[0]
+
+ ds1 = self.env.from_collection([('a', 0), ('b', 1), ('c', 2)],
+ type_info=Types.ROW([Types.STRING(), Types.INT()]))
+ ds2 = self.env.from_collection([(3, 'c'), (1, 'a'), (0, 'd')],
+ type_info=Types.ROW([Types.INT(), Types.STRING()]))
+ ds3 = ds1.connect(ds2).process(MyCoProcessFunction(), output_type=Types.STRING())
+ ds3.add_sink(self.test_sink)
+ side_sink = DataStreamTestSinkFunction()
+ ds3.get_side_output(tag).add_sink(side_sink)
+
+ self.env.execute("test_co_process_side_output")
+ main_expected = ['a', 'a', 'b', 'c', 'c', 'd']
+ self.assert_equals_sorted(main_expected, self.test_sink.get_results())
+ side_expected = ['0', '0', '1', '1', '2', '3']
+ self.assert_equals_sorted(side_expected, side_sink.get_results())
+
+ def test_keyed_process_side_output(self):
+ tag = OutputTag("side", Types.INT())
+
+ ds = self.env.from_collection([('a', 1), ('b', 2), ('a', 3), ('b', 4)],
+ type_info=Types.ROW([Types.STRING(), Types.INT()]))
+
+ class MyKeyedProcessFunction(KeyedProcessFunction):
+
+ def __init__(self):
+ self.reducing_state = None # type: ReducingState
+
+ def open(self, context: RuntimeContext):
+ self.reducing_state = context.get_reducing_state(
+ ReducingStateDescriptor("reduce", lambda i, j: i+j, Types.INT())
+ )
+
+ def process_element(self, value, ctx: 'KeyedProcessFunction.Context'):
+ yield value[1]
+ self.reducing_state.add(value[1])
+ yield tag, self.reducing_state.get()
+
+ ds2 = ds.key_by(lambda e: e[0]).process(MyKeyedProcessFunction(),
+ output_type=Types.INT())
+ main_sink = DataStreamTestSinkFunction()
+ ds2.add_sink(main_sink)
+ side_sink = DataStreamTestSinkFunction()
+ ds2.get_side_output(tag).add_sink(side_sink)
+
+ self.env.execute("test_keyed_process_side_output")
+ main_expected = ['1', '2', '3', '4']
+ self.assert_equals_sorted(main_expected, main_sink.get_results())
+ side_expected = ['1', '2', '4', '6']
+ self.assert_equals_sorted(side_expected, side_sink.get_results())
+
+ def test_keyed_co_process_side_output(self):
+ tag = OutputTag("side", Types.INT())
+
+ ds1 = self.env.from_collection([('a', 1), ('b', 2), ('a', 3), ('b', 4)],
+ type_info=Types.ROW([Types.STRING(), Types.INT()]))
+ ds2 = self.env.from_collection([(8, 'a'), (7, 'b'), (6, 'a'), (5, 'b')],
+ type_info=Types.ROW([Types.INT(), Types.STRING()]))
+
+ class MyKeyedCoProcessFunction(KeyedCoProcessFunction):
+
+ def __init__(self):
+ self.reducing_state = None # type: ReducingState
+
+ def open(self, context: RuntimeContext):
+ self.reducing_state = context.get_reducing_state(
+ ReducingStateDescriptor("reduce", lambda i, j: i+j, Types.INT())
+ )
+
+ def process_element1(self, value, ctx: 'KeyedCoProcessFunction.Context'):
+ yield value[1]
+ self.reducing_state.add(1)
+ yield tag, self.reducing_state.get()
+
+ def process_element2(self, value, ctx: 'KeyedCoProcessFunction.Context'):
+ yield value[0]
+ self.reducing_state.add(1)
+ yield tag, self.reducing_state.get()
+
+ ds3 = ds1.key_by(lambda e: e[0])\
+ .connect(ds2.key_by(lambda e: e[1]))\
+ .process(MyKeyedCoProcessFunction(), output_type=Types.INT())
+ main_sink = DataStreamTestSinkFunction()
+ ds3.add_sink(main_sink)
+ side_sink = DataStreamTestSinkFunction()
+ ds3.get_side_output(tag).add_sink(side_sink)
+
+ self.env.execute("test_keyed_co_process_side_output")
+ main_expected = ['1', '2', '3', '4', '5', '6', '7', '8']
+ self.assert_equals_sorted(main_expected, main_sink.get_results())
+ side_expected = ['1', '1', '2', '2', '3', '3', '4', '4']
+ self.assert_equals_sorted(side_expected, side_sink.get_results())
+
class StreamingModeDataStreamTests(DataStreamTests, PyFlinkStreamingTestCase):
def test_data_stream_name(self):
diff --git a/flink-python/pyflink/datastream/tests/test_window.py b/flink-python/pyflink/datastream/tests/test_window.py
index a70f3aaeaa5..9e6004d45a5 100644
--- a/flink-python/pyflink/datastream/tests/test_window.py
+++ b/flink-python/pyflink/datastream/tests/test_window.py
@@ -15,20 +15,23 @@
# See the License for the specific language governing permissions and
# limitations under the License.
################################################################################
-
from typing import Iterable, Tuple, Dict
+from pyflink.common import Configuration
from pyflink.common.time import Time
from pyflink.common.typeinfo import Types
from pyflink.common.watermark_strategy import WatermarkStrategy, TimestampAssigner
from pyflink.datastream.data_stream import DataStream
from pyflink.datastream.functions import (ProcessWindowFunction, WindowFunction, AggregateFunction)
+from pyflink.datastream.output_tag import OutputTag
from pyflink.datastream.window import (TumblingEventTimeWindows,
SlidingEventTimeWindows, EventTimeSessionWindows,
CountSlidingWindowAssigner, SessionWindowTimeGapExtractor,
CountWindow, PurgingTrigger, EventTimeTrigger, TimeWindow)
from pyflink.datastream.tests.test_util import DataStreamTestSinkFunction
+from pyflink.java_gateway import get_gateway
from pyflink.testing.test_case_utils import PyFlinkStreamingTestCase
+from pyflink.util.java_utils import get_j_env_configuration
class WindowTests(PyFlinkStreamingTestCase):
@@ -372,6 +375,41 @@ class WindowTests(PyFlinkStreamingTestCase):
expected = ['(hi,1,7,4)', '(hi,8,12,2)', '(hi,15,18,1)']
self.assert_equals_sorted(expected, results)
+ def test_side_output_late_data(self):
+ self.env.set_parallelism(1)
+ config = Configuration(
+ j_configuration=get_j_env_configuration(self.env._j_stream_execution_environment)
+ )
+ config.set_integer('python.fn-execution.bundle.size', 1)
+ jvm = get_gateway().jvm
+ watermark_strategy = WatermarkStrategy(
+ jvm.org.apache.flink.api.common.eventtime.WatermarkStrategy.forGenerator(
+ jvm.org.apache.flink.streaming.api.functions.python.eventtime.
+ PerElementWatermarkGenerator.getSupplier()
+ )
+ ).with_timestamp_assigner(SecondColumnTimestampAssigner())
+
+ tag = OutputTag('late-data', type_info=Types.ROW([Types.STRING(), Types.INT()]))
+ ds1 = self.env.from_collection([('a', 0), ('a', 8), ('a', 4), ('a', 6)],
+ type_info=Types.ROW([Types.STRING(), Types.INT()]))
+ ds2 = ds1.assign_timestamps_and_watermarks(watermark_strategy) \
+ .key_by(lambda e: e[0]) \
+ .window(TumblingEventTimeWindows.of(Time.milliseconds(5))) \
+ .allowed_lateness(0) \
+ .side_output_late_data(tag) \
+ .process(CountWindowProcessFunction(),
+ Types.TUPLE([Types.STRING(), Types.LONG(), Types.LONG(), Types.INT()]))
+ main_sink = DataStreamTestSinkFunction()
+ ds2.add_sink(main_sink)
+ side_sink = DataStreamTestSinkFunction()
+ ds2.get_side_output(tag).add_sink(side_sink)
+
+ self.env.execute('test_side_output_late_data')
+ main_expected = ['(a,0,5,1)', '(a,5,10,2)']
+ self.assert_equals_sorted(main_expected, main_sink.get_results())
+ side_expected = ['+I[a, 4]']
+ self.assert_equals_sorted(side_expected, side_sink.get_results())
+
class SecondColumnTimestampAssigner(TimestampAssigner):
diff --git a/flink-python/pyflink/datastream/window.py b/flink-python/pyflink/datastream/window.py
index c806e96335b..9da5cab96e0 100644
--- a/flink-python/pyflink/datastream/window.py
+++ b/flink-python/pyflink/datastream/window.py
@@ -19,12 +19,13 @@ import math
from abc import ABC, abstractmethod
from enum import Enum
from io import BytesIO
-from typing import TypeVar, Generic, Iterable, Collection, Any, cast
+from typing import TypeVar, Generic, Iterable, Collection, Any, cast, Optional
from pyflink.common import Time, Types
from pyflink.common.constants import MAX_LONG_VALUE, MIN_LONG_VALUE
from pyflink.common.serializer import TypeSerializer
from pyflink.datastream.functions import RuntimeContext, InternalWindowFunction, ReduceFunction
+from pyflink.datastream.output_tag import OutputTag
from pyflink.datastream.state import StateDescriptor, ReducingStateDescriptor, \
ValueStateDescriptor, State, ReducingState
from pyflink.metrics import MetricGroup
@@ -560,12 +561,14 @@ class WindowOperationDescriptor(object):
assigner: WindowAssigner,
trigger: Trigger,
allowed_lateness: int,
+ late_data_output_tag: Optional[OutputTag],
window_state_descriptor: StateDescriptor,
window_serializer: TypeSerializer,
internal_window_function: InternalWindowFunction):
self.assigner = assigner
self.trigger = trigger
self.allowed_lateness = allowed_lateness
+ self.late_data_output_tag = late_data_output_tag
self.window_state_descriptor = window_state_descriptor
self.internal_window_function = internal_window_function
self.window_serializer = window_serializer
diff --git a/flink-python/pyflink/fn_execution/beam/beam_operations_fast.pxd b/flink-python/pyflink/fn_execution/beam/beam_operations_fast.pxd
index 512f0ff1584..9808d23bd4b 100644
--- a/flink-python/pyflink/fn_execution/beam/beam_operations_fast.pxd
+++ b/flink-python/pyflink/fn_execution/beam/beam_operations_fast.pxd
@@ -46,7 +46,9 @@ cdef class IntermediateOutputProcessor(OutputProcessor):
pass
cdef class FunctionOperation(Operation):
- cdef OutputProcessor _output_processor
+ cdef dict _output_processors
+ cdef OutputProcessor _main_output_processor
+ cdef bint _has_side_output
cdef bint _is_python_coder
cdef object process_element
cdef object operation
diff --git a/flink-python/pyflink/fn_execution/beam/beam_operations_fast.pyx b/flink-python/pyflink/fn_execution/beam/beam_operations_fast.pyx
index a9ace9d1927..03dbcf2d1d9 100644
--- a/flink-python/pyflink/fn_execution/beam/beam_operations_fast.pyx
+++ b/flink-python/pyflink/fn_execution/beam/beam_operations_fast.pyx
@@ -22,14 +22,14 @@
from libc.stdint cimport *
from apache_beam.coders.coder_impl cimport OutputStream as BOutputStream
+from apache_beam.runners.worker.bundle_processor import DataOutputOperation
from apache_beam.utils cimport windowed_value
from apache_beam.utils.windowed_value cimport WindowedValue
+from pyflink.common.constants import DEFAULT_OUTPUT_TAG
from pyflink.fn_execution.coder_impl_fast cimport InputStreamWrapper
-
-from apache_beam.runners.worker.bundle_processor import DataOutputOperation
-from pyflink.fn_execution.beam.beam_coder_impl_fast import FlinkLengthPrefixCoderBeamWrapper
-from pyflink.fn_execution.table.operations import BundleOperation
+from pyflink.fn_execution.flink_fn_execution_pb2 import UserDefinedDataStreamFunction
+from pyflink.fn_execution.table.operations import BundleOperation, BaseOperation as TableOperation
from pyflink.fn_execution.profiler import Profiler
@@ -57,7 +57,7 @@ cdef class NetworkInputProcessor(InputProcessor):
cdef class IntermediateInputProcessor(InputProcessor):
def __init__(self, input_values):
- self._input_values = input_values
+ self._input_values = iter(input_values)
self._next_value = None
cpdef has_next(self):
@@ -112,19 +112,18 @@ cdef class FunctionOperation(Operation):
def __init__(self, name, spec, counter_factory, sampler, consumers, operation_cls):
super(FunctionOperation, self).__init__(name, spec, counter_factory, sampler)
- consumer = consumers['output'][0]
+ consumer = consumers[DEFAULT_OUTPUT_TAG][0]
if isinstance(consumer, DataOutputOperation):
- self._output_processor = NetworkOutputProcessor(consumer)
-
_value_coder_impl = consumer.windowed_coder.wrapped_value_coder.get_impl()._value_coder
if isinstance(_value_coder_impl, FlinkLengthPrefixCoderBeamWrapper):
self._is_python_coder = False
else:
self._is_python_coder = True
else:
- self._output_processor = IntermediateOutputProcessor(consumer)
self._is_python_coder = False
+ self._output_processors = FunctionOperation._create_output_processors(consumers) \
+ # type: Dict[str, List[OutputProcessor]]
self.operation_cls = operation_cls
self.operation = self.generate_operation()
self.process_element = self.operation.process_element
@@ -134,6 +133,14 @@ cdef class FunctionOperation(Operation):
else:
self._profiler = None
+ if isinstance(spec.serialized_fn, UserDefinedDataStreamFunction):
+ self._has_side_output = spec.serialized_fn.has_side_output
+ else:
+ # it doesn't support side output in Table API & SQL
+ self._has_side_output = False
+ if not self._has_side_output:
+ self._main_output_processor = self._output_processors[DEFAULT_OUTPUT_TAG][0]
+
cpdef start(self):
with self.scoped_start_state:
super(FunctionOperation, self).start()
@@ -150,7 +157,9 @@ cdef class FunctionOperation(Operation):
cpdef teardown(self):
with self.scoped_finish_state:
self.operation.close()
- self._output_processor.close()
+ for ps in self._output_processors.values():
+ for p in ps:
+ p.close()
cpdef process(self, WindowedValue o):
cdef InputStreamWrapper input_stream_wrapper
@@ -158,20 +167,32 @@ cdef class FunctionOperation(Operation):
with self.scoped_process_state:
if self._is_python_coder:
for value in o.value:
- self._output_processor.process_outputs(o, self.process_element(value))
+ self._main_output_processor.process_outputs(o, self.process_element(value))
else:
if isinstance(o.value, InputStreamWrapper):
input_processor = NetworkInputProcessor(o.value)
else:
input_processor = IntermediateInputProcessor(o.value)
- if isinstance(self.operation, BundleOperation):
- while input_processor.has_next():
- self.process_element(input_processor.next())
- self._output_processor.process_outputs(o, self.operation.finish_bundle())
- else:
+
+ if self._has_side_output:
while input_processor.has_next():
result = self.process_element(input_processor.next())
- self._output_processor.process_outputs(o, result)
+ for tag, row in result:
+ ps = self._output_processors.get(tag)
+ if ps is not None:
+ for p in ps:
+ (<OutputProcessor> p).process_outputs(o, [row])
+ else:
+ if isinstance(self.operation, BundleOperation):
+ while input_processor.has_next():
+ self.process_element(input_processor.next())
+ self._main_output_processor.process_outputs(o, self.operation.finish_bundle())
+ else:
+ while input_processor.has_next():
+ self._main_output_processor.process_outputs(
+ o,
+ self.process_element(input_processor.next())
+ )
def progress_metrics(self):
metrics = super(FunctionOperation, self).progress_metrics()
@@ -189,6 +210,18 @@ cdef class FunctionOperation(Operation):
"""
return self.user_monitoring_infos(transform_id)
+ @staticmethod
+ def _create_output_processors(consumers_map):
+ def _create_processor(consumer):
+ if isinstance(consumer, DataOutputOperation):
+ return NetworkOutputProcessor(consumer)
+ else:
+ return IntermediateOutputProcessor(consumer)
+
+ return {tag: [_create_processor(c) for c in consumers] for tag, consumers in
+ consumers_map.items()}
+
+
cdef object generate_operation(self):
pass
@@ -219,7 +252,12 @@ cdef class StatefulFunctionOperation(FunctionOperation):
cpdef process_timer(self, tag, timer_data):
cdef BOutputStream output_stream
- self._output_processor.process_outputs(
- self._reusable_windowed_value,
- # the field user_key holds the timer data
- self.operation.process_timer(timer_data.user_key))
+ if self._has_side_output:
+ for tag, row in self.operation.process_timer(timer_data.user_key):
+ ps = self._output_processors.get(tag)
+ if ps is not None:
+ for p in ps:
+ (<OutputProcessor> p).process_outputs(self._reusable_windowed_value, [row])
+ else:
+ self._main_output_processor.process_outputs(
+ self._reusable_windowed_value, self.operation.process_timer(timer_data.user_key))
diff --git a/flink-python/pyflink/fn_execution/beam/beam_operations_slow.py b/flink-python/pyflink/fn_execution/beam/beam_operations_slow.py
index 57a375a602e..8c85af43163 100644
--- a/flink-python/pyflink/fn_execution/beam/beam_operations_slow.py
+++ b/flink-python/pyflink/fn_execution/beam/beam_operations_slow.py
@@ -17,13 +17,15 @@
################################################################################
import abc
from abc import abstractmethod
-from typing import Iterable, Any
+from typing import Iterable, Any, Dict, List
from apache_beam.runners.worker.bundle_processor import TimerInfo, DataOutputOperation
from apache_beam.runners.worker.operations import Operation
from apache_beam.utils import windowed_value
from apache_beam.utils.windowed_value import WindowedValue
+from pyflink.common.constants import DEFAULT_OUTPUT_TAG
+from pyflink.fn_execution.flink_fn_execution_pb2 import UserDefinedDataStreamFunction
from pyflink.fn_execution.table.operations import BundleOperation
from pyflink.fn_execution.profiler import Profiler
@@ -71,11 +73,9 @@ class FunctionOperation(Operation):
def __init__(self, name, spec, counter_factory, sampler, consumers, operation_cls):
super(FunctionOperation, self).__init__(name, spec, counter_factory, sampler)
- consumer = consumers['output'][0]
- if isinstance(consumer, DataOutputOperation):
- self._output_processor = NetworkOutputProcessor(consumer)
- else:
- self._output_processor = IntermediateOutputProcessor(consumer)
+ self._output_processors = self._create_output_processors(
+ consumers
+ ) # type: Dict[str, List[OutputProcessor]]
self.operation_cls = operation_cls
self.operation = self.generate_operation()
self.process_element = self.operation.process_element
@@ -85,6 +85,14 @@ class FunctionOperation(Operation):
else:
self._profiler = None
+ if isinstance(spec.serialized_fn, UserDefinedDataStreamFunction):
+ self._has_side_output = spec.serialized_fn.has_side_output
+ else:
+ # it doesn't support side output in Table API & SQL
+ self._has_side_output = False
+ if not self._has_side_output:
+ self._main_output_processor = self._output_processors[DEFAULT_OUTPUT_TAG][0]
+
def setup(self):
super(FunctionOperation, self).setup()
@@ -110,7 +118,9 @@ class FunctionOperation(Operation):
def teardown(self):
with self.scoped_finish_state:
self.operation.close()
- self._output_processor.close()
+ for processors in self._output_processors.values():
+ for p in processors:
+ p.close()
def progress_metrics(self):
metrics = super(FunctionOperation, self).progress_metrics()
@@ -118,18 +128,27 @@ class FunctionOperation(Operation):
tag = None
receiver = self.receivers[0]
metrics.processed_elements.measured.output_element_counts[
- str(tag)] = receiver.opcounter.element_counter.value()
+ str(tag)
+ ] = receiver.opcounter.element_counter.value()
return metrics
def process(self, o: WindowedValue):
with self.scoped_process_state:
- if isinstance(self.operation, BundleOperation):
+ if self._has_side_output:
for value in o.value:
- self.process_element(value)
- self._output_processor.process_outputs(o, self.operation.finish_bundle())
+ for tag, row in self.process_element(value):
+ for p in self._output_processors.get(tag, []):
+ p.process_outputs(o, [row])
else:
- for value in o.value:
- self._output_processor.process_outputs(o, self.process_element(value))
+ if isinstance(self.operation, BundleOperation):
+ for value in o.value:
+ self.process_element(value)
+ self._main_output_processor.process_outputs(o, self.operation.finish_bundle())
+ else:
+ for value in o.value:
+ self._main_output_processor.process_outputs(
+ o, self.operation.process_element(value)
+ )
def monitoring_infos(self, transform_id, tag_to_pcollection_id):
"""
@@ -138,6 +157,19 @@ class FunctionOperation(Operation):
"""
return super().user_monitoring_infos(transform_id)
+ @staticmethod
+ def _create_output_processors(consumers_map):
+ def _create_processor(consumer):
+ if isinstance(consumer, DataOutputOperation):
+ return NetworkOutputProcessor(consumer)
+ else:
+ return IntermediateOutputProcessor(consumer)
+
+ return {
+ tag: [_create_processor(c) for c in consumers]
+ for tag, consumers in consumers_map.items()
+ }
+
@abstractmethod
def generate_operation(self):
pass
@@ -146,19 +178,29 @@ class FunctionOperation(Operation):
class StatelessFunctionOperation(FunctionOperation):
def __init__(self, name, spec, counter_factory, sampler, consumers, operation_cls):
super(StatelessFunctionOperation, self).__init__(
- name, spec, counter_factory, sampler, consumers, operation_cls)
+ name, spec, counter_factory, sampler, consumers, operation_cls
+ )
def generate_operation(self):
return self.operation_cls(self.spec.serialized_fn)
class StatefulFunctionOperation(FunctionOperation):
- def __init__(self, name, spec, counter_factory, sampler, consumers, operation_cls,
- keyed_state_backend):
+ def __init__(
+ self,
+ name,
+ spec,
+ counter_factory,
+ sampler,
+ consumers,
+ operation_cls,
+ keyed_state_backend,
+ ):
self._keyed_state_backend = keyed_state_backend
self._reusable_windowed_value = windowed_value.create(None, -1, None, None)
super(StatefulFunctionOperation, self).__init__(
- name, spec, counter_factory, sampler, consumers, operation_cls)
+ name, spec, counter_factory, sampler, consumers, operation_cls
+ )
def generate_operation(self):
return self.operation_cls(self.spec.serialized_fn, self._keyed_state_backend)
@@ -168,7 +210,14 @@ class StatefulFunctionOperation(FunctionOperation):
self.operation.add_timer_info(timer_info)
def process_timer(self, tag, timer_data):
- self._output_processor.process_outputs(
- self._reusable_windowed_value,
+ if self._has_side_output:
# the field user_key holds the timer data
- self.operation.process_timer(timer_data.user_key))
+ for tag, row in self.operation.process_timer(timer_data.user_key):
+ for p in self._output_processors.get(tag, []):
+ p.process_outputs(self._reusable_windowed_value, [row])
+ else:
+ self._main_output_processor.process_outputs(
+ self._reusable_windowed_value,
+ # the field user_key holds the timer data
+ self.operation.process_timer(timer_data.user_key),
+ )
diff --git a/flink-python/pyflink/fn_execution/datastream/input_handler.py b/flink-python/pyflink/fn_execution/datastream/input_handler.py
index 6722ba6ad01..f18e72fa84c 100644
--- a/flink-python/pyflink/fn_execution/datastream/input_handler.py
+++ b/flink-python/pyflink/fn_execution/datastream/input_handler.py
@@ -16,10 +16,12 @@
# limitations under the License.
################################################################################
from abc import ABC
-from collections import Iterable
from enum import Enum
+from typing import cast, Iterable
from pyflink.common import Row
+from pyflink.common.constants import DEFAULT_OUTPUT_TAG
+from pyflink.datastream.output_tag import OutputTag
from pyflink.fn_execution.datastream.timerservice_impl import InternalTimerServiceImpl
@@ -33,18 +35,25 @@ class RunnerInputHandler(ABC):
Handler which handles normal input data.
"""
- def __init__(self,
- internal_timer_service: InternalTimerServiceImpl,
- process_element_func):
+ def __init__(
+ self,
+ internal_timer_service: InternalTimerServiceImpl,
+ process_element_func,
+ has_side_output: bool
+ ):
self._internal_timer_service = internal_timer_service
self._process_element_func = process_element_func
+ self._has_side_output = has_side_output
def process_element(self, value) -> Iterable:
timestamp = value[0]
watermark = value[1]
data = value[2]
self._advance_watermark(watermark)
- yield from _emit_results(timestamp, watermark, self._process_element_func(data, timestamp))
+ yield from _emit_results(timestamp,
+ watermark,
+ self._process_element_func(data, timestamp),
+ self._has_side_output)
def _advance_watermark(self, watermark: int) -> None:
self._internal_timer_service.advance_watermark(watermark)
@@ -55,15 +64,19 @@ class TimerHandler(ABC):
Handler which handles normal input data.
"""
- def __init__(self,
- internal_timer_service: InternalTimerServiceImpl,
- on_event_time_func,
- on_processing_time_func,
- namespace_coder):
+ def __init__(
+ self,
+ internal_timer_service: InternalTimerServiceImpl,
+ on_event_time_func,
+ on_processing_time_func,
+ namespace_coder,
+ has_side_output
+ ):
self._internal_timer_service = internal_timer_service
self._on_event_time_func = on_event_time_func
self._on_processing_time_func = on_processing_time_func
self._namespace_coder = namespace_coder
+ self._has_side_output = has_side_output
def process_timer(self, timer_data) -> Iterable:
timer_type = timer_data[0]
@@ -79,10 +92,18 @@ class TimerHandler(ABC):
namespace = None
if timer_type == TimerType.EVENT_TIME.value:
yield from _emit_results(
- timestamp, watermark, self._on_event_time(timestamp, key, namespace))
+ timestamp,
+ watermark,
+ self._on_event_time(timestamp, key, namespace),
+ self._has_side_output
+ )
elif timer_type == TimerType.PROCESSING_TIME.value:
yield from _emit_results(
- timestamp, watermark, self._on_processing_time(timestamp, key, namespace))
+ timestamp,
+ watermark,
+ self._on_processing_time(timestamp, key, namespace),
+ self._has_side_output
+ )
else:
raise Exception("Unsupported timer type: %d" % timer_type)
@@ -96,7 +117,16 @@ class TimerHandler(ABC):
self._internal_timer_service.advance_watermark(watermark)
-def _emit_results(timestamp, watermark, results):
+def _emit_results(timestamp, watermark, results, has_side_output):
if results:
- for result in results:
- yield Row(timestamp, watermark, result)
+ if has_side_output:
+ for result in results:
+ if isinstance(result, tuple) and isinstance(result[0], OutputTag):
+ yield cast(OutputTag, result[0]).tag_id, Row(
+ timestamp, watermark, result[1]
+ )
+ else:
+ yield DEFAULT_OUTPUT_TAG, Row(timestamp, watermark, result)
+ else:
+ for result in results:
+ yield Row(timestamp, watermark, result)
diff --git a/flink-python/pyflink/fn_execution/datastream/operations.py b/flink-python/pyflink/fn_execution/datastream/operations.py
index afaed5a732d..c13db669764 100644
--- a/flink-python/pyflink/fn_execution/datastream/operations.py
+++ b/flink-python/pyflink/fn_execution/datastream/operations.py
@@ -20,25 +20,32 @@ import abc
from pyflink.common import Row
from pyflink.common.serializer import VoidNamespaceSerializer
from pyflink.datastream import TimeDomain, RuntimeContext
+from pyflink.datastream.window import WindowOperationDescriptor
from pyflink.fn_execution import pickle
-from pyflink.fn_execution.datastream.process_function import \
- InternalKeyedProcessFunctionOnTimerContext, InternalKeyedProcessFunctionContext, \
- InternalProcessFunctionContext
+from pyflink.fn_execution.datastream.process_function import (
+ InternalKeyedProcessFunctionOnTimerContext,
+ InternalKeyedProcessFunctionContext,
+ InternalProcessFunctionContext,
+)
from pyflink.fn_execution.datastream.runtime_context import StreamingRuntimeContext
from pyflink.fn_execution.datastream.window.window_operator import WindowOperator
from pyflink.fn_execution.datastream.timerservice_impl import (
- TimerServiceImpl, InternalTimerServiceImpl, NonKeyedTimerServiceImpl)
-from pyflink.fn_execution.datastream.input_handler import (RunnerInputHandler, TimerHandler,
- _emit_results)
+ TimerServiceImpl,
+ InternalTimerServiceImpl,
+ NonKeyedTimerServiceImpl,
+)
+from pyflink.fn_execution.datastream.input_handler import (
+ RunnerInputHandler,
+ TimerHandler,
+ _emit_results,
+)
from pyflink.metrics.metricbase import GenericMetricGroup
-
DATA_STREAM_STATELESS_FUNCTION_URN = "flink:transform:ds:stateless_function:v1"
DATA_STREAM_STATEFUL_FUNCTION_URN = "flink:transform:ds:stateful_function:v1"
class Operation(abc.ABC):
-
def __init__(self, serialized_fn):
if serialized_fn.metric_enabled:
self.base_metric_group = GenericMetricGroup(None, None)
@@ -68,15 +75,18 @@ class Operation(abc.ABC):
class StatelessOperation(Operation):
-
def __init__(self, serialized_fn):
super(StatelessOperation, self).__init__(serialized_fn)
- self.open_func, self.close_func, self.process_element_func = \
- extract_stateless_function(
- user_defined_function_proto=serialized_fn,
- runtime_context=StreamingRuntimeContext.of(
- serialized_fn.runtime_context,
- self.base_metric_group))
+ (
+ self.open_func,
+ self.close_func,
+ self.process_element_func,
+ ) = extract_stateless_function(
+ user_defined_function_proto=serialized_fn,
+ runtime_context=StreamingRuntimeContext.of(
+ serialized_fn.runtime_context, self.base_metric_group
+ ),
+ )
def open(self):
self.open_func()
@@ -89,19 +99,24 @@ class StatelessOperation(Operation):
class StatefulOperation(Operation):
-
def __init__(self, serialized_fn, keyed_state_backend):
super(StatefulOperation, self).__init__(serialized_fn)
self.keyed_state_backend = keyed_state_backend
- self.open_func, self.close_func, self.process_element_func, self.process_timer_func, \
- self.internal_timer_service = \
- extract_stateful_function(
- user_defined_function_proto=serialized_fn,
- runtime_context=StreamingRuntimeContext.of(
- serialized_fn.runtime_context,
- self.base_metric_group,
- self.keyed_state_backend),
- keyed_state_backend=self.keyed_state_backend)
+ (
+ self.open_func,
+ self.close_func,
+ self.process_element_func,
+ self.process_timer_func,
+ self.internal_timer_service,
+ ) = extract_stateful_function(
+ user_defined_function_proto=serialized_fn,
+ runtime_context=StreamingRuntimeContext.of(
+ serialized_fn.runtime_context,
+ self.base_metric_group,
+ self.keyed_state_backend,
+ ),
+ keyed_state_backend=self.keyed_state_backend,
+ )
def finish(self):
super().finish()
@@ -123,7 +138,9 @@ class StatefulOperation(Operation):
self.internal_timer_service.add_timer_info(timer_info)
-def extract_stateless_function(user_defined_function_proto, runtime_context: RuntimeContext):
+def extract_stateless_function(
+ user_defined_function_proto, runtime_context: RuntimeContext
+):
"""
Extracts user-defined-function from the proto representation of a
:class:`Function`.
@@ -134,9 +151,11 @@ def extract_stateless_function(user_defined_function_proto, runtime_context: Run
from pyflink.fn_execution import flink_fn_execution_pb2
func_type = user_defined_function_proto.function_type
+ has_side_output = user_defined_function_proto.has_side_output
UserDefinedDataStreamFunction = flink_fn_execution_pb2.UserDefinedDataStreamFunction
if func_type == UserDefinedDataStreamFunction.REVISE_OUTPUT:
+
def open_func():
pass
@@ -173,7 +192,7 @@ def extract_stateless_function(user_defined_function_proto, runtime_context: Run
ctx.set_timestamp(timestamp)
ctx.timer_service().advance_watermark(watermark)
results = process_element(value[2], ctx)
- yield from _emit_results(timestamp, watermark, results)
+ yield from _emit_results(timestamp, watermark, results, has_side_output)
process_element_func = wrapped_func
@@ -195,7 +214,7 @@ def extract_stateless_function(user_defined_function_proto, runtime_context: Run
else:
results = process_element2(normal_data[2], ctx)
- yield from _emit_results(timestamp, watermark, results)
+ yield from _emit_results(timestamp, watermark, results, has_side_output)
process_element_func = wrapped_func
@@ -205,13 +224,14 @@ def extract_stateless_function(user_defined_function_proto, runtime_context: Run
return open_func, close_func, process_element_func
-def extract_stateful_function(user_defined_function_proto,
- runtime_context: RuntimeContext,
- keyed_state_backend):
+def extract_stateful_function(
+ user_defined_function_proto, runtime_context: RuntimeContext, keyed_state_backend
+):
from pyflink.fn_execution import flink_fn_execution_pb2
func_type = user_defined_function_proto.function_type
user_defined_func = pickle.loads(user_defined_function_proto.payload)
+ has_side_output = user_defined_function_proto.has_side_output
internal_timer_service = InternalTimerServiceImpl(keyed_state_backend)
def state_key_selector(normal_data):
@@ -224,8 +244,10 @@ def extract_stateful_function(user_defined_function_proto,
return normal_data[1]
UserDefinedDataStreamFunction = flink_fn_execution_pb2.UserDefinedDataStreamFunction
- if func_type in (UserDefinedDataStreamFunction.KEYED_PROCESS,
- UserDefinedDataStreamFunction.KEYED_CO_PROCESS):
+ if func_type in (
+ UserDefinedDataStreamFunction.KEYED_PROCESS,
+ UserDefinedDataStreamFunction.KEYED_CO_PROCESS,
+ ):
timer_service = TimerServiceImpl(internal_timer_service)
ctx = InternalKeyedProcessFunctionContext(timer_service)
on_timer_ctx = InternalKeyedProcessFunctionOnTimerContext(timer_service)
@@ -263,7 +285,9 @@ def extract_stateful_function(user_defined_function_proto,
ctx.set_timestamp(timestamp)
ctx.set_current_key(user_key_selector(normal_data))
keyed_state_backend.set_current_key(state_key_selector(normal_data))
- return process_function.process_element(input_selector(normal_data), ctx)
+ return process_function.process_element(
+ input_selector(normal_data), ctx
+ )
elif func_type == UserDefinedDataStreamFunction.KEYED_CO_PROCESS:
@@ -279,18 +303,25 @@ def extract_stateful_function(user_defined_function_proto,
keyed_state_backend.set_current_key(state_key_selector(user_input))
if is_left:
- return process_function.process_element1(input_selector(user_input), ctx)
+ return process_function.process_element1(
+ input_selector(user_input), ctx
+ )
else:
- return process_function.process_element2(input_selector(user_input), ctx)
+ return process_function.process_element2(
+ input_selector(user_input), ctx
+ )
else:
raise Exception("Unsupported func_type: " + str(func_type))
elif func_type == UserDefinedDataStreamFunction.WINDOW:
- window_operation_descriptor = user_defined_func
+ window_operation_descriptor = (
+ user_defined_func
+ ) # type: WindowOperationDescriptor
window_assigner = window_operation_descriptor.assigner
window_trigger = window_operation_descriptor.trigger
allowed_lateness = window_operation_descriptor.allowed_lateness
+ late_data_output_tag = window_operation_descriptor.late_data_output_tag
window_state_descriptor = window_operation_descriptor.window_state_descriptor
internal_window_function = window_operation_descriptor.internal_window_function
window_serializer = window_operation_descriptor.window_serializer
@@ -304,7 +335,9 @@ def extract_stateful_function(user_defined_function_proto,
window_state_descriptor,
internal_window_function,
window_trigger,
- allowed_lateness)
+ allowed_lateness,
+ late_data_output_tag,
+ )
internal_timer_service.set_namespace_serializer(window_serializer)
def open_func():
@@ -315,7 +348,9 @@ def extract_stateful_function(user_defined_function_proto,
def process_element(normal_data, timestamp: int):
keyed_state_backend.set_current_key(state_key_selector(normal_data))
- return window_operator.process_element(input_selector(normal_data), timestamp)
+ return window_operator.process_element(
+ input_selector(normal_data), timestamp
+ )
def on_event_time(timestamp: int, key, namespace):
keyed_state_backend.set_current_key(key)
@@ -328,16 +363,22 @@ def extract_stateful_function(user_defined_function_proto,
else:
raise Exception("Unsupported function_type: " + str(func_type))
- input_handler = RunnerInputHandler(
- internal_timer_service,
- process_element)
+ input_handler = RunnerInputHandler(internal_timer_service, process_element, has_side_output)
process_element_func = input_handler.process_element
timer_handler = TimerHandler(
internal_timer_service,
on_event_time,
on_processing_time,
- keyed_state_backend._namespace_coder_impl)
+ keyed_state_backend._namespace_coder_impl,
+ has_side_output
+ )
process_timer_func = timer_handler.process_timer
- return open_func, close_func, process_element_func, process_timer_func, internal_timer_service
+ return (
+ open_func,
+ close_func,
+ process_element_func,
+ process_timer_func,
+ internal_timer_service,
+ )
diff --git a/flink-python/pyflink/fn_execution/datastream/window/window_operator.py b/flink-python/pyflink/fn_execution/datastream/window/window_operator.py
index 234bc97863b..a5ceb7b6ab5 100644
--- a/flink-python/pyflink/fn_execution/datastream/window/window_operator.py
+++ b/flink-python/pyflink/fn_execution/datastream/window/window_operator.py
@@ -16,11 +16,12 @@
# limitations under the License.
################################################################################
import typing
-from typing import TypeVar, Iterable, Collection
+from typing import TypeVar, Iterable, Collection, Optional
from pyflink.common.constants import MAX_LONG_VALUE
from pyflink.datastream import WindowAssigner, Trigger, MergingWindowAssigner, TriggerResult
from pyflink.datastream.functions import KeyedStateStore, RuntimeContext, InternalWindowFunction
+from pyflink.datastream.output_tag import OutputTag
from pyflink.datastream.state import StateDescriptor, ListStateDescriptor, \
ReducingStateDescriptor, AggregatingStateDescriptor, ValueStateDescriptor, MapStateDescriptor, \
State, AggregatingState, ReducingState, MapState, ListState, ValueState, AppendingState
@@ -273,7 +274,8 @@ class WindowOperator(object):
window_state_descriptor: StateDescriptor,
window_function: InternalWindowFunction,
trigger: Trigger,
- allowed_lateness: int):
+ allowed_lateness: int,
+ late_data_output_tag: Optional[OutputTag]):
self.window_assigner = window_assigner
self.keyed_state_backend = keyed_state_backend
self.user_key_selector = user_key_selector
@@ -281,6 +283,7 @@ class WindowOperator(object):
self.window_function = window_function
self.trigger = trigger
self.allowed_lateness = allowed_lateness
+ self.late_data_output_tag = late_data_output_tag
self.num_late_records_dropped = None
self.internal_timer_service = None # type: InternalTimerService
@@ -408,7 +411,10 @@ class WindowOperator(object):
self.register_cleanup_timer(window)
if is_skipped_element and self.is_element_late(value, timestamp):
- self.num_late_records_dropped.inc()
+ if self.late_data_output_tag is not None:
+ yield self.late_data_output_tag, value
+ else:
+ self.num_late_records_dropped.inc()
def on_event_time(self, timestamp, key, namespace) -> None:
self.trigger_context.user_key = self.user_key_selector(key)
diff --git a/flink-python/pyflink/fn_execution/flink_fn_execution_pb2.py b/flink-python/pyflink/fn_execution/flink_fn_execution_pb2.py
index 90b0bc5af53..80cbaab505d 100644
--- a/flink-python/pyflink/fn_execution/flink_fn_execution_pb2.py
+++ b/flink-python/pyflink/fn_execution/flink_fn_execution_pb2.py
@@ -36,7 +36,7 @@ DESCRIPTOR = _descriptor.FileDescriptor(
name='flink-fn-execution.proto',
package='org.apache.flink.fn_execution.v1',
syntax='proto3',
- serialized_pb=_b('\n\x18\x66link-fn-execution.proto\x12 org.apache.flink.fn_execution.v1\"\x86\x01\n\x05Input\x12\x44\n\x03udf\x18\x01 \x01(\x0b\x32\x35.org.apache.flink.fn_execution.v1.UserDefinedFunctionH\x00\x12\x15\n\x0binputOffset\x18\x02 \x01(\x05H\x00\x12\x17\n\rinputConstant\x18\x03 \x01(\x0cH\x00\x42\x07\n\x05input\"\xa8\x01\n\x13UserDefinedFunction\x12\x0f\n\x07payload\x18\x01 \x01(\x0c\x12\x37\n\x06inputs\x18\x02 \x03(\x0b\x32\'.org.apache.flink.fn_execution.v1.Input\x12\x14 [...]
+ serialized_pb=_b('\n\x18\x66link-fn-execution.proto\x12 org.apache.flink.fn_execution.v1\"\x86\x01\n\x05Input\x12\x44\n\x03udf\x18\x01 \x01(\x0b\x32\x35.org.apache.flink.fn_execution.v1.UserDefinedFunctionH\x00\x12\x15\n\x0binputOffset\x18\x02 \x01(\x05H\x00\x12\x17\n\rinputConstant\x18\x03 \x01(\x0cH\x00\x42\x07\n\x05input\"\xa8\x01\n\x13UserDefinedFunction\x12\x0f\n\x07payload\x18\x01 \x01(\x0c\x12\x37\n\x06inputs\x18\x02 \x03(\x0b\x32\'.org.apache.flink.fn_execution.v1.Input\x12\x14 [...]
)
@@ -380,8 +380,8 @@ _USERDEFINEDDATASTREAMFUNCTION_FUNCTIONTYPE = _descriptor.EnumDescriptor(
],
containing_type=None,
options=None,
- serialized_start=6836,
- serialized_end=6951,
+ serialized_start=6861,
+ serialized_end=6976,
)
_sym_db.RegisterEnumDescriptor(_USERDEFINEDDATASTREAMFUNCTION_FUNCTIONTYPE)
@@ -406,8 +406,8 @@ _STATEDESCRIPTOR_STATETTLCONFIG_CLEANUPSTRATEGIES_STRATEGIES = _descriptor.EnumD
],
containing_type=None,
options=None,
- serialized_start=8513,
- serialized_end=8611,
+ serialized_start=8538,
+ serialized_end=8636,
)
_sym_db.RegisterEnumDescriptor(_STATEDESCRIPTOR_STATETTLCONFIG_CLEANUPSTRATEGIES_STRATEGIES)
@@ -424,8 +424,8 @@ _STATEDESCRIPTOR_STATETTLCONFIG_CLEANUPSTRATEGIES_EMPTYCLEANUPSTRATEGY = _descri
],
containing_type=None,
options=None,
- serialized_start=8613,
- serialized_end=8655,
+ serialized_start=8638,
+ serialized_end=8680,
)
_sym_db.RegisterEnumDescriptor(_STATEDESCRIPTOR_STATETTLCONFIG_CLEANUPSTRATEGIES_EMPTYCLEANUPSTRATEGY)
@@ -450,8 +450,8 @@ _STATEDESCRIPTOR_STATETTLCONFIG_UPDATETYPE = _descriptor.EnumDescriptor(
],
containing_type=None,
options=None,
- serialized_start=8657,
- serialized_end=8725,
+ serialized_start=8682,
+ serialized_end=8750,
)
_sym_db.RegisterEnumDescriptor(_STATEDESCRIPTOR_STATETTLCONFIG_UPDATETYPE)
@@ -472,8 +472,8 @@ _STATEDESCRIPTOR_STATETTLCONFIG_STATEVISIBILITY = _descriptor.EnumDescriptor(
],
containing_type=None,
options=None,
- serialized_start=8727,
- serialized_end=8801,
+ serialized_start=8752,
+ serialized_end=8826,
)
_sym_db.RegisterEnumDescriptor(_STATEDESCRIPTOR_STATETTLCONFIG_STATEVISIBILITY)
@@ -490,8 +490,8 @@ _STATEDESCRIPTOR_STATETTLCONFIG_TTLTIMECHARACTERISTIC = _descriptor.EnumDescript
],
containing_type=None,
options=None,
- serialized_start=8803,
- serialized_end=8846,
+ serialized_start=8828,
+ serialized_end=8871,
)
_sym_db.RegisterEnumDescriptor(_STATEDESCRIPTOR_STATETTLCONFIG_TTLTIMECHARACTERISTIC)
@@ -512,8 +512,8 @@ _CODERINFODESCRIPTOR_MODE = _descriptor.EnumDescriptor(
],
containing_type=None,
options=None,
- serialized_start=9813,
- serialized_end=9845,
+ serialized_start=9838,
+ serialized_end=9870,
)
_sym_db.RegisterEnumDescriptor(_CODERINFODESCRIPTOR_MODE)
@@ -1865,8 +1865,8 @@ _USERDEFINEDDATASTREAMFUNCTION_JOBPARAMETER = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=6453,
- serialized_end=6495,
+ serialized_start=6478,
+ serialized_end=6520,
)
_USERDEFINEDDATASTREAMFUNCTION_RUNTIMECONTEXT = _descriptor.Descriptor(
@@ -1944,8 +1944,8 @@ _USERDEFINEDDATASTREAMFUNCTION_RUNTIMECONTEXT = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=6498,
- serialized_end=6834,
+ serialized_start=6523,
+ serialized_end=6859,
)
_USERDEFINEDDATASTREAMFUNCTION = _descriptor.Descriptor(
@@ -1998,26 +1998,33 @@ _USERDEFINEDDATASTREAMFUNCTION = _descriptor.Descriptor(
is_extension=False, extension_scope=None,
options=None, file=DESCRIPTOR),
_descriptor.FieldDescriptor(
- name='state_cache_size', full_name='org.apache.flink.fn_execution.v1.UserDefinedDataStreamFunction.state_cache_size', index=6,
- number=7, type=5, cpp_type=1, label=1,
- has_default_value=False, default_value=0,
+ name='has_side_output', full_name='org.apache.flink.fn_execution.v1.UserDefinedDataStreamFunction.has_side_output', index=6,
+ number=7, type=8, cpp_type=7, label=1,
+ has_default_value=False, default_value=False,
message_type=None, enum_type=None, containing_type=None,
is_extension=False, extension_scope=None,
options=None, file=DESCRIPTOR),
_descriptor.FieldDescriptor(
- name='map_state_read_cache_size', full_name='org.apache.flink.fn_execution.v1.UserDefinedDataStreamFunction.map_state_read_cache_size', index=7,
+ name='state_cache_size', full_name='org.apache.flink.fn_execution.v1.UserDefinedDataStreamFunction.state_cache_size', index=7,
number=8, type=5, cpp_type=1, label=1,
has_default_value=False, default_value=0,
message_type=None, enum_type=None, containing_type=None,
is_extension=False, extension_scope=None,
options=None, file=DESCRIPTOR),
_descriptor.FieldDescriptor(
- name='map_state_write_cache_size', full_name='org.apache.flink.fn_execution.v1.UserDefinedDataStreamFunction.map_state_write_cache_size', index=8,
+ name='map_state_read_cache_size', full_name='org.apache.flink.fn_execution.v1.UserDefinedDataStreamFunction.map_state_read_cache_size', index=8,
number=9, type=5, cpp_type=1, label=1,
has_default_value=False, default_value=0,
message_type=None, enum_type=None, containing_type=None,
is_extension=False, extension_scope=None,
options=None, file=DESCRIPTOR),
+ _descriptor.FieldDescriptor(
+ name='map_state_write_cache_size', full_name='org.apache.flink.fn_execution.v1.UserDefinedDataStreamFunction.map_state_write_cache_size', index=9,
+ number=10, type=5, cpp_type=1, label=1,
+ has_default_value=False, default_value=0,
+ message_type=None, enum_type=None, containing_type=None,
+ is_extension=False, extension_scope=None,
+ options=None, file=DESCRIPTOR),
],
extensions=[
],
@@ -2032,7 +2039,7 @@ _USERDEFINEDDATASTREAMFUNCTION = _descriptor.Descriptor(
oneofs=[
],
serialized_start=5984,
- serialized_end=6951,
+ serialized_end=6976,
)
@@ -2069,8 +2076,8 @@ _STATEDESCRIPTOR_STATETTLCONFIG_CLEANUPSTRATEGIES_INCREMENTALCLEANUPSTRATEGY = _
extension_ranges=[],
oneofs=[
],
- serialized_start=7735,
- serialized_end=7823,
+ serialized_start=7760,
+ serialized_end=7848,
)
_STATEDESCRIPTOR_STATETTLCONFIG_CLEANUPSTRATEGIES_ROCKSDBCOMPACTFILTERCLEANUPSTRATEGY = _descriptor.Descriptor(
@@ -2099,8 +2106,8 @@ _STATEDESCRIPTOR_STATETTLCONFIG_CLEANUPSTRATEGIES_ROCKSDBCOMPACTFILTERCLEANUPSTR
extension_ranges=[],
oneofs=[
],
- serialized_start=7825,
- serialized_end=7900,
+ serialized_start=7850,
+ serialized_end=7925,
)
_STATEDESCRIPTOR_STATETTLCONFIG_CLEANUPSTRATEGIES_MAPSTRATEGIESENTRY = _descriptor.Descriptor(
@@ -2153,8 +2160,8 @@ _STATEDESCRIPTOR_STATETTLCONFIG_CLEANUPSTRATEGIES_MAPSTRATEGIESENTRY = _descript
name='CleanupStrategy', full_name='org.apache.flink.fn_execution.v1.StateDescriptor.StateTTLConfig.CleanupStrategies.MapStrategiesEntry.CleanupStrategy',
index=0, containing_type=None, fields=[]),
],
- serialized_start=7903,
- serialized_end=8511,
+ serialized_start=7928,
+ serialized_end=8536,
)
_STATEDESCRIPTOR_STATETTLCONFIG_CLEANUPSTRATEGIES = _descriptor.Descriptor(
@@ -2192,8 +2199,8 @@ _STATEDESCRIPTOR_STATETTLCONFIG_CLEANUPSTRATEGIES = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=7557,
- serialized_end=8655,
+ serialized_start=7582,
+ serialized_end=8680,
)
_STATEDESCRIPTOR_STATETTLCONFIG = _descriptor.Descriptor(
@@ -2253,8 +2260,8 @@ _STATEDESCRIPTOR_STATETTLCONFIG = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=7086,
- serialized_end=8846,
+ serialized_start=7111,
+ serialized_end=8871,
)
_STATEDESCRIPTOR = _descriptor.Descriptor(
@@ -2290,8 +2297,8 @@ _STATEDESCRIPTOR = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=6954,
- serialized_end=8846,
+ serialized_start=6979,
+ serialized_end=8871,
)
@@ -2321,8 +2328,8 @@ _CODERINFODESCRIPTOR_FLATTENROWTYPE = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=9442,
- serialized_end=9516,
+ serialized_start=9467,
+ serialized_end=9541,
)
_CODERINFODESCRIPTOR_ROWTYPE = _descriptor.Descriptor(
@@ -2351,8 +2358,8 @@ _CODERINFODESCRIPTOR_ROWTYPE = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=9518,
- serialized_end=9585,
+ serialized_start=9543,
+ serialized_end=9610,
)
_CODERINFODESCRIPTOR_ARROWTYPE = _descriptor.Descriptor(
@@ -2381,8 +2388,8 @@ _CODERINFODESCRIPTOR_ARROWTYPE = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=9587,
- serialized_end=9656,
+ serialized_start=9612,
+ serialized_end=9681,
)
_CODERINFODESCRIPTOR_OVERWINDOWARROWTYPE = _descriptor.Descriptor(
@@ -2411,8 +2418,8 @@ _CODERINFODESCRIPTOR_OVERWINDOWARROWTYPE = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=9658,
- serialized_end=9737,
+ serialized_start=9683,
+ serialized_end=9762,
)
_CODERINFODESCRIPTOR_RAWTYPE = _descriptor.Descriptor(
@@ -2441,8 +2448,8 @@ _CODERINFODESCRIPTOR_RAWTYPE = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=9739,
- serialized_end=9811,
+ serialized_start=9764,
+ serialized_end=9836,
)
_CODERINFODESCRIPTOR = _descriptor.Descriptor(
@@ -2517,8 +2524,8 @@ _CODERINFODESCRIPTOR = _descriptor.Descriptor(
name='data_type', full_name='org.apache.flink.fn_execution.v1.CoderInfoDescriptor.data_type',
index=0, containing_type=None, fields=[]),
],
- serialized_start=8849,
- serialized_end=9858,
+ serialized_start=8874,
+ serialized_end=9883,
)
_INPUT.fields_by_name['udf'].message_type = _USERDEFINEDFUNCTION
diff --git a/flink-python/pyflink/proto/flink-fn-execution.proto b/flink-python/pyflink/proto/flink-fn-execution.proto
index 7ea69a55f8b..5b8ba5e9b42 100644
--- a/flink-python/pyflink/proto/flink-fn-execution.proto
+++ b/flink-python/pyflink/proto/flink-fn-execution.proto
@@ -371,13 +371,14 @@ message UserDefinedDataStreamFunction {
bool metric_enabled = 4;
TypeInfo key_type_info = 5;
bool profile_enabled = 6;
+ bool has_side_output = 7;
// The state cache size.
- int32 state_cache_size = 7;
+ int32 state_cache_size = 8;
// The map state read cache size.
- int32 map_state_read_cache_size = 8;
+ int32 map_state_read_cache_size = 9;
// The map_state_write_cache_size.
- int32 map_state_write_cache_size = 9;
+ int32 map_state_write_cache_size = 10;
}
// A representation of State
diff --git a/flink-python/src/main/java/org/apache/flink/python/Constants.java b/flink-python/src/main/java/org/apache/flink/python/Constants.java
index 082716615e2..80b957684d3 100644
--- a/flink-python/src/main/java/org/apache/flink/python/Constants.java
+++ b/flink-python/src/main/java/org/apache/flink/python/Constants.java
@@ -31,12 +31,13 @@ public class Constants {
// execution graph
public static final String TRANSFORM_ID = "transform";
public static final String MAIN_INPUT_NAME = "input";
- public static final String MAIN_OUTPUT_NAME = "output";
+ public static final String MAIN_OUTPUT_NAME = "";
public static final String WINDOW_STRATEGY = "windowing_strategy";
public static final String INPUT_COLLECTION_ID = "input";
- public static final String OUTPUT_COLLECTION_ID = "output";
+ public static final String OUTPUT_COLLECTION_ID = "";
+ public static final String SIDE_OUTPUT_CODER_PREFIX = "side_coder-";
public static final String WINDOW_CODER_ID = "window_coder";
public static final String TIMER_ID = "timer";
public static final String TIMER_CODER_ID = "timer_coder";
diff --git a/flink-python/src/main/java/org/apache/flink/python/PythonFunctionRunner.java b/flink-python/src/main/java/org/apache/flink/python/PythonFunctionRunner.java
index 4a4174a768d..078778a3371 100644
--- a/flink-python/src/main/java/org/apache/flink/python/PythonFunctionRunner.java
+++ b/flink-python/src/main/java/org/apache/flink/python/PythonFunctionRunner.java
@@ -19,7 +19,7 @@
package org.apache.flink.python;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.ReadableConfig;
/** The base interface of runner which is responsible for the execution of Python functions. */
@@ -51,7 +51,7 @@ public interface PythonFunctionRunner {
* empty. f0 means the byte array buffer which stores the Python function result. f1 means
* the length of the Python function result byte array.
*/
- Tuple2<byte[], Integer> pollResult() throws Exception;
+ Tuple3<String, byte[], Integer> pollResult() throws Exception;
/**
* Retrieves the Python function result, waiting if necessary until an element becomes
@@ -61,7 +61,7 @@ public interface PythonFunctionRunner {
* stores the Python function result. f1 means the length of the Python function result byte
* array.
*/
- Tuple2<byte[], Integer> takeResult() throws Exception;
+ Tuple3<String, byte[], Integer> takeResult() throws Exception;
/**
* Forces to finish the processing of the current bundle of elements. It will flush the data
diff --git a/flink-python/src/main/java/org/apache/flink/python/chain/PythonOperatorChainingOptimizer.java b/flink-python/src/main/java/org/apache/flink/python/chain/PythonOperatorChainingOptimizer.java
index e573be20772..bbd0d19a513 100644
--- a/flink-python/src/main/java/org/apache/flink/python/chain/PythonOperatorChainingOptimizer.java
+++ b/flink-python/src/main/java/org/apache/flink/python/chain/PythonOperatorChainingOptimizer.java
@@ -67,6 +67,8 @@ import java.util.Map;
import java.util.Queue;
import java.util.Set;
+import static org.apache.flink.python.util.PythonConfigUtil.getOperatorFactory;
+
/**
* A util class which attempts to chain all available Python functions.
*
@@ -282,6 +284,7 @@ public class PythonOperatorChainingOptimizer {
final AbstractDataStreamPythonFunctionOperator<?> chainedOperator =
upOperator.copy(downPythonFunctionInfo, downOperator.getProducedType());
+ chainedOperator.addSideOutputTags(downOperator.getSideOutputTags());
// set partition custom flag
chainedOperator.setContainsPartitionCustom(
@@ -454,18 +457,6 @@ public class PythonOperatorChainingOptimizer {
// ----------------------- Utility Methods -----------------------
- private static StreamOperatorFactory<?> getOperatorFactory(Transformation<?> transform) {
- if (transform instanceof OneInputTransformation) {
- return ((OneInputTransformation<?, ?>) transform).getOperatorFactory();
- } else if (transform instanceof TwoInputTransformation) {
- return ((TwoInputTransformation<?, ?, ?>) transform).getOperatorFactory();
- } else if (transform instanceof AbstractMultipleInputTransformation) {
- return ((AbstractMultipleInputTransformation<?>) transform).getOperatorFactory();
- } else {
- return null;
- }
- }
-
private static void replaceInput(
Transformation<?> transformation,
Transformation<?> oldInput,
diff --git a/flink-python/src/main/java/org/apache/flink/python/util/PythonConfigUtil.java b/flink-python/src/main/java/org/apache/flink/python/util/PythonConfigUtil.java
index 3c6085b7254..8ae55cc8760 100644
--- a/flink-python/src/main/java/org/apache/flink/python/util/PythonConfigUtil.java
+++ b/flink-python/src/main/java/org/apache/flink/python/util/PythonConfigUtil.java
@@ -33,9 +33,12 @@ import org.apache.flink.streaming.api.operators.python.AbstractPythonFunctionOpe
import org.apache.flink.streaming.api.transformations.AbstractMultipleInputTransformation;
import org.apache.flink.streaming.api.transformations.OneInputTransformation;
import org.apache.flink.streaming.api.transformations.PartitionTransformation;
+import org.apache.flink.streaming.api.transformations.SideOutputTransformation;
import org.apache.flink.streaming.api.transformations.TwoInputTransformation;
import org.apache.flink.streaming.runtime.partitioner.ForwardPartitioner;
+import org.apache.flink.util.OutputTag;
+import org.apache.flink.shaded.guava30.com.google.common.collect.Iterables;
import org.apache.flink.shaded.guava30.com.google.common.collect.Queues;
import org.apache.flink.shaded.guava30.com.google.common.collect.Sets;
@@ -98,6 +101,8 @@ public class PythonConfigUtil {
}
}
}
+
+ processSideOutput(env.getTransformations());
}
/** Extract the configurations which is used in the Python operators. */
@@ -110,6 +115,52 @@ public class PythonConfigUtil {
return pythonConfig.toConfiguration();
}
+ /**
+ * Process {@link SideOutputTransformation}s, set the {@link OutputTag}s into the Python
+ * corresponding operator to make it aware of the {@link OutputTag}s.
+ */
+ private static void processSideOutput(List<Transformation<?>> transformations) {
+ final Set<Transformation<?>> visitedTransforms = Sets.newIdentityHashSet();
+ final Queue<Transformation<?>> queue = Queues.newArrayDeque(transformations);
+
+ while (!queue.isEmpty()) {
+ Transformation<?> transform = queue.poll();
+ visitedTransforms.add(transform);
+
+ if (transform instanceof SideOutputTransformation) {
+ final SideOutputTransformation<?> sideTransform =
+ (SideOutputTransformation<?>) transform;
+ final Transformation<?> upTransform =
+ Iterables.getOnlyElement(sideTransform.getInputs());
+ if (PythonConfigUtil.isPythonDataStreamOperator(upTransform)) {
+ final AbstractDataStreamPythonFunctionOperator<?> upOperator =
+ (AbstractDataStreamPythonFunctionOperator<?>)
+ ((SimpleOperatorFactory<?>) getOperatorFactory(upTransform))
+ .getOperator();
+ upOperator.addSideOutputTag(sideTransform.getOutputTag());
+ }
+ }
+
+ for (Transformation<?> upTransform : transform.getInputs()) {
+ if (!visitedTransforms.contains(upTransform)) {
+ queue.add(upTransform);
+ }
+ }
+ }
+ }
+
+ public static StreamOperatorFactory<?> getOperatorFactory(Transformation<?> transform) {
+ if (transform instanceof OneInputTransformation) {
+ return ((OneInputTransformation<?, ?>) transform).getOperatorFactory();
+ } else if (transform instanceof TwoInputTransformation) {
+ return ((TwoInputTransformation<?, ?, ?>) transform).getOperatorFactory();
+ } else if (transform instanceof AbstractMultipleInputTransformation) {
+ return ((AbstractMultipleInputTransformation<?>) transform).getOperatorFactory();
+ } else {
+ return null;
+ }
+ }
+
/**
* Configure the {@link AbstractOneInputPythonFunctionOperator} to be chained with the
* upstream/downstream operator by setting their parallelism, slot sharing group, co-location
diff --git a/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractDataStreamPythonFunctionOperator.java b/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractDataStreamPythonFunctionOperator.java
index 6685abb5556..d72b03024b5 100644
--- a/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractDataStreamPythonFunctionOperator.java
+++ b/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractDataStreamPythonFunctionOperator.java
@@ -20,17 +20,26 @@ package org.apache.flink.streaming.api.operators.python;
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.fnexecution.v1.FlinkFnApi;
import org.apache.flink.streaming.api.functions.python.DataStreamPythonFunctionInfo;
+import org.apache.flink.streaming.api.utils.PythonTypeUtils;
import org.apache.flink.table.functions.python.PythonEnv;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.OutputTag;
import org.apache.flink.util.Preconditions;
import javax.annotation.Nullable;
+import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
+import static org.apache.flink.streaming.api.utils.ProtoUtils.createRawTypeCoderInfoDescriptorProto;
+
/** Base class for all Python DataStream operators. */
@Internal
public abstract class AbstractDataStreamPythonFunctionOperator<OUT>
@@ -55,6 +64,10 @@ public abstract class AbstractDataStreamPythonFunctionOperator<OUT>
/** The TypeInformation of output data. */
private final TypeInformation<OUT> outputTypeInfo;
+ private final Map<String, OutputTag<?>> sideOutputTags;
+
+ private transient Map<String, TypeSerializer<Row>> sideOutputSerializers;
+
public AbstractDataStreamPythonFunctionOperator(
Configuration config,
DataStreamPythonFunctionInfo pythonFunctionInfo,
@@ -62,6 +75,19 @@ public abstract class AbstractDataStreamPythonFunctionOperator<OUT>
super(config);
this.pythonFunctionInfo = Preconditions.checkNotNull(pythonFunctionInfo);
this.outputTypeInfo = Preconditions.checkNotNull(outputTypeInfo);
+ this.sideOutputTags = new HashMap<>();
+ }
+
+ @Override
+ public void open() throws Exception {
+ sideOutputSerializers = new HashMap<>();
+ for (Map.Entry<String, OutputTag<?>> entry : sideOutputTags.entrySet()) {
+ sideOutputSerializers.put(
+ entry.getKey(),
+ PythonTypeUtils.TypeInfoToSerializerConverter.typeInfoSerializerConverter(
+ getSideOutputTypeInfo(entry.getValue())));
+ }
+ super.open();
}
@Override
@@ -97,6 +123,49 @@ public abstract class AbstractDataStreamPythonFunctionOperator<OUT>
return this.containsPartitionCustom;
}
+ // ----------------------------------------------------------------------
+ // Side outputs
+ // ----------------------------------------------------------------------
+
+ public void addSideOutputTag(OutputTag<?> outputTag) {
+ sideOutputTags.put(outputTag.getId(), outputTag);
+ }
+
+ protected OutputTag<?> getOutputTagById(String id) {
+ Preconditions.checkArgument(sideOutputTags.containsKey(id));
+ return sideOutputTags.get(id);
+ }
+
+ public void addSideOutputTags(Collection<OutputTag<?>> outputTags) {
+ outputTags.forEach(this::addSideOutputTag);
+ }
+
+ public Collection<OutputTag<?>> getSideOutputTags() {
+ return sideOutputTags.values();
+ }
+
+ protected Map<String, FlinkFnApi.CoderInfoDescriptor> createSideOutputCoderDescriptors() {
+ Map<String, FlinkFnApi.CoderInfoDescriptor> descriptorMap = new HashMap<>();
+ for (Map.Entry<String, OutputTag<?>> entry : sideOutputTags.entrySet()) {
+ descriptorMap.put(
+ entry.getKey(),
+ createRawTypeCoderInfoDescriptorProto(
+ getSideOutputTypeInfo(entry.getValue()),
+ FlinkFnApi.CoderInfoDescriptor.Mode.MULTIPLE,
+ false));
+ }
+ return descriptorMap;
+ }
+
+ protected TypeSerializer<Row> getSideOutputTypeSerializerById(String id) {
+ Preconditions.checkArgument(sideOutputSerializers.containsKey(id));
+ return sideOutputSerializers.get(id);
+ }
+
+ private static TypeInformation<Row> getSideOutputTypeInfo(OutputTag<?> outputTag) {
+ return Types.ROW(Types.LONG, outputTag.getTypeInfo());
+ }
+
// ----------------------------------------------------------------------
// Getters
// ----------------------------------------------------------------------
diff --git a/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractExternalPythonFunctionOperator.java b/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractExternalPythonFunctionOperator.java
index d95ffafb5af..0a72db1936c 100644
--- a/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractExternalPythonFunctionOperator.java
+++ b/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractExternalPythonFunctionOperator.java
@@ -19,7 +19,7 @@
package org.apache.flink.streaming.api.operators.python;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.python.PythonFunctionRunner;
import org.apache.flink.python.env.PythonDependencyInfo;
@@ -92,10 +92,10 @@ public abstract class AbstractExternalPythonFunctionOperator<OUT>
((BeamPythonFunctionRunner) pythonFunctionRunner).notifyNoMoreResults();
}
});
- Tuple2<byte[], Integer> resultTuple;
+ Tuple3<String, byte[], Integer> resultTuple;
while (!flushThreadFinish.get()) {
resultTuple = pythonFunctionRunner.takeResult();
- if (resultTuple.f1 != 0) {
+ if (resultTuple.f2 != 0) {
emitResult(resultTuple);
emitResults();
}
@@ -136,14 +136,14 @@ public abstract class AbstractExternalPythonFunctionOperator<OUT>
}
protected void emitResults() throws Exception {
- Tuple2<byte[], Integer> resultTuple;
- while ((resultTuple = pythonFunctionRunner.pollResult()) != null && resultTuple.f1 != 0) {
+ Tuple3<String, byte[], Integer> resultTuple;
+ while ((resultTuple = pythonFunctionRunner.pollResult()) != null && resultTuple.f2 != 0) {
emitResult(resultTuple);
}
}
/** Sends the execution result to the downstream operator. */
- public abstract void emitResult(Tuple2<byte[], Integer> resultTuple) throws Exception;
+ public abstract void emitResult(Tuple3<String, byte[], Integer> resultTuple) throws Exception;
/**
* Creates the {@link PythonFunctionRunner} which is responsible for Python user-defined
diff --git a/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractOneInputPythonFunctionOperator.java b/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractOneInputPythonFunctionOperator.java
index 76202fbbbda..663a1bd9c34 100644
--- a/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractOneInputPythonFunctionOperator.java
+++ b/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractOneInputPythonFunctionOperator.java
@@ -22,7 +22,7 @@ import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.memory.ByteArrayInputStreamWithPos;
import org.apache.flink.core.memory.ByteArrayOutputStreamWithPos;
@@ -38,6 +38,7 @@ import org.apache.flink.streaming.api.utils.PythonTypeUtils;
import org.apache.flink.types.Row;
import org.apache.flink.util.Preconditions;
+import static org.apache.flink.python.Constants.OUTPUT_COLLECTION_ID;
import static org.apache.flink.streaming.api.utils.ProtoUtils.createRawTypeCoderInfoDescriptorProto;
/**
@@ -122,12 +123,18 @@ public abstract class AbstractOneInputPythonFunctionOperator<IN, OUT>
}
@Override
- public void emitResult(Tuple2<byte[], Integer> resultTuple) throws Exception {
- byte[] rawResult = resultTuple.f0;
- int length = resultTuple.f1;
+ public void emitResult(Tuple3<String, byte[], Integer> resultTuple) throws Exception {
+ String tag = resultTuple.f0;
+ byte[] rawResult = resultTuple.f1;
+ int length = resultTuple.f2;
bais.setBuffer(rawResult, 0, length);
- Row runnerOutput = runnerOutputTypeSerializer.deserialize(baisWrapper);
- runnerOutputCollector.collect(runnerOutput);
+ if (tag.equals(OUTPUT_COLLECTION_ID)) {
+ Row runnerOutput = runnerOutputTypeSerializer.deserialize(baisWrapper);
+ runnerOutputCollector.collect(runnerOutput);
+ } else {
+ Row runnerOutput = getSideOutputTypeSerializerById(tag).deserialize(baisWrapper);
+ runnerOutputCollector.collect(getOutputTagById(tag), runnerOutput);
+ }
}
public void processElement(long timestamp, long watermark, Object element) throws Exception {
diff --git a/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractTwoInputPythonFunctionOperator.java b/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractTwoInputPythonFunctionOperator.java
index 733fe2132cf..5d096ecb6d5 100644
--- a/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractTwoInputPythonFunctionOperator.java
+++ b/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractTwoInputPythonFunctionOperator.java
@@ -22,7 +22,7 @@ import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.memory.ByteArrayInputStreamWithPos;
@@ -38,6 +38,7 @@ import org.apache.flink.streaming.api.operators.python.collector.RunnerOutputCol
import org.apache.flink.types.Row;
import org.apache.flink.util.Preconditions;
+import static org.apache.flink.python.Constants.OUTPUT_COLLECTION_ID;
import static org.apache.flink.streaming.api.utils.ProtoUtils.createRawTypeCoderInfoDescriptorProto;
import static org.apache.flink.streaming.api.utils.PythonTypeUtils.TypeInfoToSerializerConverter.typeInfoSerializerConverter;
@@ -113,12 +114,18 @@ public abstract class AbstractTwoInputPythonFunctionOperator<IN1, IN2, OUT>
}
@Override
- public void emitResult(Tuple2<byte[], Integer> resultTuple) throws Exception {
- byte[] rawResult = resultTuple.f0;
- int length = resultTuple.f1;
+ public void emitResult(Tuple3<String, byte[], Integer> resultTuple) throws Exception {
+ String tag = resultTuple.f0;
+ byte[] rawResult = resultTuple.f1;
+ int length = resultTuple.f2;
bais.setBuffer(rawResult, 0, length);
- Row runnerOutput = runnerOutputTypeSerializer.deserialize(baisWrapper);
- runnerOutputCollector.collect(runnerOutput);
+ if (tag.equals(OUTPUT_COLLECTION_ID)) {
+ Row runnerOutput = runnerOutputTypeSerializer.deserialize(baisWrapper);
+ runnerOutputCollector.collect(runnerOutput);
+ } else {
+ Row runnerOutput = getSideOutputTypeSerializerById(tag).deserialize(baisWrapper);
+ runnerOutputCollector.collect(getOutputTagById(tag), runnerOutput);
+ }
}
public void processElement(boolean isLeft, long timestamp, long watermark, Object element)
diff --git a/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/PythonCoProcessOperator.java b/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/PythonCoProcessOperator.java
index 3a13f6725dd..c6005d8a106 100644
--- a/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/PythonCoProcessOperator.java
+++ b/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/PythonCoProcessOperator.java
@@ -81,6 +81,7 @@ public class PythonCoProcessOperator<IN1, IN2, OUT>
inBatchExecutionMode(getKeyedStateBackend()),
config.get(PYTHON_METRIC_ENABLED),
config.get(PYTHON_PROFILE_ENABLED),
+ getSideOutputTags().size() > 0,
config.get(STATE_CACHE_SIZE),
config.get(MAP_STATE_READ_CACHE_SIZE),
config.get(MAP_STATE_WRITE_CACHE_SIZE)),
@@ -103,7 +104,8 @@ public class PythonCoProcessOperator<IN1, IN2, OUT>
.asClassLoader()),
createInputCoderInfoDescriptor(),
createOutputCoderInfoDescriptor(),
- null);
+ null,
+ createSideOutputCoderDescriptors());
}
@Override
diff --git a/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/PythonKeyedCoProcessOperator.java b/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/PythonKeyedCoProcessOperator.java
index 26ccaedf9ad..08314636cf8 100644
--- a/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/PythonKeyedCoProcessOperator.java
+++ b/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/PythonKeyedCoProcessOperator.java
@@ -120,6 +120,7 @@ public class PythonKeyedCoProcessOperator<OUT>
inBatchExecutionMode(getKeyedStateBackend()),
config.get(PYTHON_METRIC_ENABLED),
config.get(PYTHON_PROFILE_ENABLED),
+ getSideOutputTags().size() > 0,
config.get(STATE_CACHE_SIZE),
config.get(MAP_STATE_READ_CACHE_SIZE),
config.get(MAP_STATE_WRITE_CACHE_SIZE)),
@@ -148,7 +149,8 @@ public class PythonKeyedCoProcessOperator<OUT>
.asClassLoader()),
createInputCoderInfoDescriptor(),
createOutputCoderInfoDescriptor(),
- createTimerDataCoderInfoDescriptorProto(timerDataTypeInfo));
+ createTimerDataCoderInfoDescriptorProto(timerDataTypeInfo),
+ createSideOutputCoderDescriptors());
}
@Override
diff --git a/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/PythonKeyedProcessOperator.java b/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/PythonKeyedProcessOperator.java
index 2896ed3be47..3671b7fce61 100644
--- a/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/PythonKeyedProcessOperator.java
+++ b/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/PythonKeyedProcessOperator.java
@@ -150,6 +150,7 @@ public class PythonKeyedProcessOperator<OUT>
inBatchExecutionMode(getKeyedStateBackend()),
config.get(PYTHON_METRIC_ENABLED),
config.get(PYTHON_PROFILE_ENABLED),
+ getSideOutputTags().size() > 0,
config.get(STATE_CACHE_SIZE),
config.get(MAP_STATE_READ_CACHE_SIZE),
config.get(MAP_STATE_WRITE_CACHE_SIZE)),
@@ -177,7 +178,8 @@ public class PythonKeyedProcessOperator<OUT>
.asClassLoader()),
createInputCoderInfoDescriptor(),
createOutputCoderInfoDescriptor(),
- createTimerDataCoderInfoDescriptorProto(timerDataTypeInfo));
+ createTimerDataCoderInfoDescriptorProto(timerDataTypeInfo),
+ createSideOutputCoderDescriptors());
}
@Override
diff --git a/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/PythonProcessOperator.java b/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/PythonProcessOperator.java
index 48e11fab123..31b20bd6fe6 100644
--- a/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/PythonProcessOperator.java
+++ b/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/PythonProcessOperator.java
@@ -78,6 +78,7 @@ public class PythonProcessOperator<IN, OUT>
inBatchExecutionMode(getKeyedStateBackend()),
config.get(PYTHON_METRIC_ENABLED),
config.get(PYTHON_PROFILE_ENABLED),
+ getSideOutputTags().size() > 0,
config.get(STATE_CACHE_SIZE),
config.get(MAP_STATE_READ_CACHE_SIZE),
config.get(MAP_STATE_WRITE_CACHE_SIZE)),
@@ -100,7 +101,8 @@ public class PythonProcessOperator<IN, OUT>
.asClassLoader()),
createInputCoderInfoDescriptor(),
createOutputCoderInfoDescriptor(),
- null);
+ null,
+ createSideOutputCoderDescriptors());
}
@Override
diff --git a/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/collector/RunnerOutputCollector.java b/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/collector/RunnerOutputCollector.java
index e72ec23f164..7aafc10cb36 100644
--- a/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/collector/RunnerOutputCollector.java
+++ b/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/collector/RunnerOutputCollector.java
@@ -22,8 +22,10 @@ import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.operators.TimestampedCollector;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;
+import org.apache.flink.util.OutputTag;
/** Output collector for Python UDF runner. */
@Internal
@@ -31,8 +33,11 @@ public final class RunnerOutputCollector<OUT> implements Collector<Row> {
private final TimestampedCollector<OUT> collector;
+ private final StreamRecord<?> reuse;
+
public RunnerOutputCollector(TimestampedCollector<OUT> collector) {
this.collector = collector;
+ this.reuse = new StreamRecord<>(null);
}
@SuppressWarnings("unchecked")
@@ -47,6 +52,20 @@ public final class RunnerOutputCollector<OUT> implements Collector<Row> {
collector.collect((OUT) runnerOutput.getField(1));
}
+ @SuppressWarnings("unchecked")
+ public <X> void collect(OutputTag<X> outputTag, Row runnerOutput) {
+ long ts = (long) runnerOutput.getField(0);
+ StreamRecord<X> typedReuse = (StreamRecord<X>) reuse;
+ typedReuse.replace((X) runnerOutput.getField(1));
+ if (ts != Long.MIN_VALUE) {
+ reuse.setTimestamp(ts);
+ collector.collect(outputTag, typedReuse);
+ } else {
+ reuse.eraseTimestamp();
+ collector.collect(outputTag, typedReuse);
+ }
+ }
+
@Override
public void close() {
collector.close();
diff --git a/flink-python/src/main/java/org/apache/flink/streaming/api/runners/python/beam/BeamDataStreamPythonFunctionRunner.java b/flink-python/src/main/java/org/apache/flink/streaming/api/runners/python/beam/BeamDataStreamPythonFunctionRunner.java
index df3a651b8d1..5ae7e9c2d0a 100644
--- a/flink-python/src/main/java/org/apache/flink/streaming/api/runners/python/beam/BeamDataStreamPythonFunctionRunner.java
+++ b/flink-python/src/main/java/org/apache/flink/streaming/api/runners/python/beam/BeamDataStreamPythonFunctionRunner.java
@@ -36,7 +36,9 @@ import org.apache.beam.runners.core.construction.graph.TimerReference;
import javax.annotation.Nullable;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
import static org.apache.flink.python.Constants.INPUT_COLLECTION_ID;
@@ -79,7 +81,8 @@ public class BeamDataStreamPythonFunctionRunner extends BeamPythonFunctionRunner
double managedMemoryFraction,
FlinkFnApi.CoderInfoDescriptor inputCoderDescriptor,
FlinkFnApi.CoderInfoDescriptor outputCoderDescriptor,
- FlinkFnApi.CoderInfoDescriptor timerCoderDescriptor) {
+ @Nullable FlinkFnApi.CoderInfoDescriptor timerCoderDescriptor,
+ Map<String, FlinkFnApi.CoderInfoDescriptor> sideOutputCoderDescriptors) {
super(
taskName,
environmentManager,
@@ -91,7 +94,8 @@ public class BeamDataStreamPythonFunctionRunner extends BeamPythonFunctionRunner
memoryManager,
managedMemoryFraction,
inputCoderDescriptor,
- outputCoderDescriptor);
+ outputCoderDescriptor,
+ sideOutputCoderDescriptors);
this.headOperatorFunctionUrn = Preconditions.checkNotNull(headOperatorFunctionUrn);
Preconditions.checkArgument(
userDefinedDataStreamFunctions != null
@@ -102,85 +106,144 @@ public class BeamDataStreamPythonFunctionRunner extends BeamPythonFunctionRunner
@Override
protected void buildTransforms(RunnerApi.Components.Builder componentsBuilder) {
- for (int i = 0; i < userDefinedDataStreamFunctions.size() + 1; i++) {
- String functionUrn;
- if (i == 0) {
- functionUrn = headOperatorFunctionUrn;
- } else {
- functionUrn = STATELESS_FUNCTION_URN;
- }
+ for (int i = 0; i < userDefinedDataStreamFunctions.size(); i++) {
+ final Map<String, String> outputCollectionMap = new HashMap<>();
- FlinkFnApi.UserDefinedDataStreamFunction functionProto;
- if (i < userDefinedDataStreamFunctions.size()) {
- functionProto = userDefinedDataStreamFunctions.get(i);
- } else {
- // the last function in the operation tree is used to prune the watermark column
- functionProto = createReviseOutputDataStreamFunctionProto();
+ // Prepare side outputs
+ if (i == userDefinedDataStreamFunctions.size() - 1) {
+ for (Map.Entry<String, FlinkFnApi.CoderInfoDescriptor> entry :
+ sideOutputCoderDescriptors.entrySet()) {
+ final String reviseCollectionId =
+ COLLECTION_PREFIX + "revise-" + entry.getKey();
+ final String reviseCoderId = CODER_PREFIX + "revise-" + entry.getKey();
+ outputCollectionMap.put(entry.getKey(), reviseCollectionId);
+ addCollectionToComponents(componentsBuilder, reviseCollectionId, reviseCoderId);
+ }
}
- // Use ParDoPayload as a wrapper of the actual payload as timer is only supported in
- // ParDo
- final RunnerApi.ParDoPayload.Builder payloadBuilder =
- RunnerApi.ParDoPayload.newBuilder()
- .setDoFn(
- RunnerApi.FunctionSpec.newBuilder()
- .setUrn(functionUrn)
- .setPayload(
- org.apache.beam.vendor.grpc.v1p26p0.com.google
- .protobuf.ByteString.copyFrom(
- functionProto.toByteArray()))
- .build());
-
- // Timer is only available in the head operator
- if (i == 0 && timerCoderDescriptor != null) {
- payloadBuilder.putTimerFamilySpecs(
- TIMER_ID,
- RunnerApi.TimerFamilySpec.newBuilder()
- // this field is not used, always set it as event time
- .setTimeDomain(RunnerApi.TimeDomain.Enum.EVENT_TIME)
- .setTimerFamilyCoderId(WRAPPER_TIMER_CODER_ID)
- .build());
- }
+ // Prepare main outputs
+ final String outputCollectionId = COLLECTION_PREFIX + i;
+ final String outputCoderId = CODER_PREFIX + i;
+ outputCollectionMap.put(MAIN_OUTPUT_NAME, outputCollectionId);
+ addCollectionToComponents(componentsBuilder, outputCollectionId, outputCoderId);
- final String transformName = TRANSFORM_ID_PREFIX + i;
-
- final RunnerApi.PTransform.Builder transformBuilder =
- RunnerApi.PTransform.newBuilder()
- .setUniqueName(transformName)
- .setSpec(
- RunnerApi.FunctionSpec.newBuilder()
- .setUrn(
- BeamUrns.getUrn(
- RunnerApi.StandardPTransforms.Primitives
- .PAR_DO))
- .setPayload(payloadBuilder.build().toByteString())
- .build());
-
- // prepare inputs
+ final String transformId = TRANSFORM_ID_PREFIX + i;
+ final FlinkFnApi.UserDefinedDataStreamFunction functionProto =
+ userDefinedDataStreamFunctions.get(i);
if (i == 0) {
- transformBuilder.putInputs(MAIN_INPUT_NAME, INPUT_COLLECTION_ID);
+ addTransformToComponents(
+ componentsBuilder,
+ transformId,
+ createUdfPayload(functionProto, headOperatorFunctionUrn, true),
+ INPUT_COLLECTION_ID,
+ outputCollectionMap);
} else {
- transformBuilder.putInputs(MAIN_INPUT_NAME, COLLECTION_PREFIX + (i - 1));
+ addTransformToComponents(
+ componentsBuilder,
+ transformId,
+ createUdfPayload(functionProto, STATELESS_FUNCTION_URN, false),
+ COLLECTION_PREFIX + (i - 1),
+ outputCollectionMap);
}
+ }
- // prepare outputs
- if (i == userDefinedDataStreamFunctions.size()) {
- transformBuilder.putOutputs(MAIN_OUTPUT_NAME, OUTPUT_COLLECTION_ID);
- } else {
- transformBuilder.putOutputs(MAIN_OUTPUT_NAME, COLLECTION_PREFIX + i);
-
- componentsBuilder
- .putPcollections(
- COLLECTION_PREFIX + i,
- RunnerApi.PCollection.newBuilder()
- .setWindowingStrategyId(WINDOW_STRATEGY)
- .setCoderId(CODER_PREFIX + i)
- .build())
- .putCoders(CODER_PREFIX + i, createCoderProto(inputCoderDescriptor));
- }
+ // Add REVISE_OUTPUT transformation for side outputs
+ for (Map.Entry<String, FlinkFnApi.CoderInfoDescriptor> entry :
+ sideOutputCoderDescriptors.entrySet()) {
+ addTransformToComponents(
+ componentsBuilder,
+ TRANSFORM_ID_PREFIX + "revise-" + entry.getKey(),
+ createRevisePayload(),
+ COLLECTION_PREFIX + "revise-" + entry.getKey(),
+ Collections.singletonMap(MAIN_OUTPUT_NAME, entry.getKey()));
+ }
- componentsBuilder.putTransforms(transformName, transformBuilder.build());
+ // Add REVISE_OUTPUT transformation for main output
+ addTransformToComponents(
+ componentsBuilder,
+ TRANSFORM_ID_PREFIX + "revise",
+ createRevisePayload(),
+ COLLECTION_PREFIX + (userDefinedDataStreamFunctions.size() - 1),
+ Collections.singletonMap(MAIN_OUTPUT_NAME, OUTPUT_COLLECTION_ID));
+ }
+
+ private RunnerApi.ParDoPayload createRevisePayload() {
+ final FlinkFnApi.UserDefinedDataStreamFunction proto =
+ createReviseOutputDataStreamFunctionProto();
+ final RunnerApi.ParDoPayload.Builder payloadBuilder =
+ RunnerApi.ParDoPayload.newBuilder()
+ .setDoFn(
+ RunnerApi.FunctionSpec.newBuilder()
+ .setUrn(STATELESS_FUNCTION_URN)
+ .setPayload(
+ org.apache.beam.vendor.grpc.v1p26p0.com.google
+ .protobuf.ByteString.copyFrom(
+ proto.toByteArray()))
+ .build());
+ return payloadBuilder.build();
+ }
+
+ private RunnerApi.ParDoPayload createUdfPayload(
+ FlinkFnApi.UserDefinedDataStreamFunction proto, String urn, boolean createTimer) {
+ // Use ParDoPayload as a wrapper of the actual payload as timer is only supported in
+ // ParDo
+ final RunnerApi.ParDoPayload.Builder payloadBuilder =
+ RunnerApi.ParDoPayload.newBuilder()
+ .setDoFn(
+ RunnerApi.FunctionSpec.newBuilder()
+ .setUrn(urn)
+ .setPayload(
+ org.apache.beam.vendor.grpc.v1p26p0.com.google
+ .protobuf.ByteString.copyFrom(
+ proto.toByteArray()))
+ .build());
+
+ // Timer is only available in the head operator
+ if (createTimer && timerCoderDescriptor != null) {
+ payloadBuilder.putTimerFamilySpecs(
+ TIMER_ID,
+ RunnerApi.TimerFamilySpec.newBuilder()
+ // this field is not used, always set it as event time
+ .setTimeDomain(RunnerApi.TimeDomain.Enum.EVENT_TIME)
+ .setTimerFamilyCoderId(WRAPPER_TIMER_CODER_ID)
+ .build());
}
+
+ return payloadBuilder.build();
+ }
+
+ private void addTransformToComponents(
+ RunnerApi.Components.Builder componentsBuilder,
+ String transformId,
+ RunnerApi.ParDoPayload payload,
+ String inputCollectionId,
+ Map<String, String> outputCollectionMap) {
+ final RunnerApi.PTransform.Builder transformBuilder =
+ RunnerApi.PTransform.newBuilder()
+ .setUniqueName(transformId)
+ .setSpec(
+ RunnerApi.FunctionSpec.newBuilder()
+ .setUrn(
+ BeamUrns.getUrn(
+ RunnerApi.StandardPTransforms.Primitives
+ .PAR_DO))
+ .setPayload(payload.toByteString())
+ .build());
+ transformBuilder.putInputs(MAIN_INPUT_NAME, inputCollectionId);
+ transformBuilder.putAllOutputs(outputCollectionMap);
+ componentsBuilder.putTransforms(transformId, transformBuilder.build());
+ }
+
+ private void addCollectionToComponents(
+ RunnerApi.Components.Builder componentsBuilder, String collectionId, String coderId) {
+ componentsBuilder
+ .putPcollections(
+ collectionId,
+ RunnerApi.PCollection.newBuilder()
+ .setWindowingStrategyId(WINDOW_STRATEGY)
+ .setCoderId(coderId)
+ .build())
+ .putCoders(coderId, createCoderProto(inputCoderDescriptor));
}
@Override
diff --git a/flink-python/src/main/java/org/apache/flink/streaming/api/runners/python/beam/BeamPythonFunctionRunner.java b/flink-python/src/main/java/org/apache/flink/streaming/api/runners/python/beam/BeamPythonFunctionRunner.java
index ad04d9b5299..d39375ccf43 100644
--- a/flink-python/src/main/java/org/apache/flink/streaming/api/runners/python/beam/BeamPythonFunctionRunner.java
+++ b/flink-python/src/main/java/org/apache/flink/streaming/api/runners/python/beam/BeamPythonFunctionRunner.java
@@ -22,6 +22,7 @@ import org.apache.flink.annotation.Internal;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.fnexecution.v1.FlinkFnApi;
import org.apache.flink.python.PythonFunctionRunner;
@@ -77,7 +78,7 @@ import javax.annotation.Nullable;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
-import java.util.Arrays;
+import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
@@ -90,6 +91,7 @@ import java.util.stream.Collectors;
import static org.apache.beam.runners.core.construction.BeamUrns.getUrn;
import static org.apache.flink.python.Constants.INPUT_COLLECTION_ID;
import static org.apache.flink.python.Constants.OUTPUT_COLLECTION_ID;
+import static org.apache.flink.python.Constants.SIDE_OUTPUT_CODER_PREFIX;
import static org.apache.flink.python.Constants.TIMER_CODER_ID;
import static org.apache.flink.python.Constants.WINDOW_CODER_ID;
import static org.apache.flink.python.Constants.WINDOW_STRATEGY;
@@ -131,6 +133,7 @@ public abstract class BeamPythonFunctionRunner implements PythonFunctionRunner {
protected final FlinkFnApi.CoderInfoDescriptor inputCoderDescriptor;
protected final FlinkFnApi.CoderInfoDescriptor outputCoderDescriptor;
+ protected final Map<String, FlinkFnApi.CoderInfoDescriptor> sideOutputCoderDescriptors;
// ------------------------------------------------------------------------
@@ -164,11 +167,11 @@ public abstract class BeamPythonFunctionRunner implements PythonFunctionRunner {
*/
private transient RemoteBundle remoteBundle;
- /** The Python function execution result tuple: (raw bytes, length). */
- private transient Tuple2<byte[], Integer> reusableResultTuple;
+ /** The Python function execution result tuple: (output tag, raw bytes, length). */
+ private transient Tuple3<String, byte[], Integer> reusableResultTuple;
/** Buffers the Python function execution result which has still not been processed. */
- @VisibleForTesting protected transient LinkedBlockingQueue<byte[]> resultBuffer;
+ @VisibleForTesting protected transient LinkedBlockingQueue<Tuple2<String, byte[]>> resultBuffer;
/** The receiver which forwards the input elements to a remote environment for processing. */
@VisibleForTesting protected transient FnDataReceiver<WindowedValue<byte[]>> mainInputReceiver;
@@ -190,7 +193,8 @@ public abstract class BeamPythonFunctionRunner implements PythonFunctionRunner {
MemoryManager memoryManager,
double managedMemoryFraction,
FlinkFnApi.CoderInfoDescriptor inputCoderDescriptor,
- FlinkFnApi.CoderInfoDescriptor outputCoderDescriptor) {
+ FlinkFnApi.CoderInfoDescriptor outputCoderDescriptor,
+ Map<String, FlinkFnApi.CoderInfoDescriptor> sideOutputCoderDescriptors) {
this.taskName = Preconditions.checkNotNull(taskName);
this.environmentManager = Preconditions.checkNotNull(environmentManager);
this.flinkMetricContainer = flinkMetricContainer;
@@ -202,6 +206,7 @@ public abstract class BeamPythonFunctionRunner implements PythonFunctionRunner {
this.managedMemoryFraction = managedMemoryFraction;
this.inputCoderDescriptor = Preconditions.checkNotNull(inputCoderDescriptor);
this.outputCoderDescriptor = Preconditions.checkNotNull(outputCoderDescriptor);
+ this.sideOutputCoderDescriptors = Preconditions.checkNotNull(sideOutputCoderDescriptors);
}
// ------------------------------------------------------------------------
@@ -210,7 +215,7 @@ public abstract class BeamPythonFunctionRunner implements PythonFunctionRunner {
public void open(ReadableConfig config) throws Exception {
this.bundleStarted = false;
this.resultBuffer = new LinkedBlockingQueue<>();
- this.reusableResultTuple = new Tuple2<>();
+ this.reusableResultTuple = new Tuple3<>();
stateRequestHandler =
getStateRequestHandler(
@@ -340,23 +345,24 @@ public abstract class BeamPythonFunctionRunner implements PythonFunctionRunner {
}
@Override
- public Tuple2<byte[], Integer> pollResult() throws Exception {
- byte[] result = resultBuffer.poll();
+ public Tuple3<String, byte[], Integer> pollResult() throws Exception {
+ Tuple2<String, byte[]> result = resultBuffer.poll();
if (result == null) {
return null;
- } else {
- this.reusableResultTuple.f0 = result;
- this.reusableResultTuple.f1 = result.length;
- return this.reusableResultTuple;
}
+ reusableResultTuple.f0 = result.f0;
+ reusableResultTuple.f1 = result.f1;
+ reusableResultTuple.f2 = result.f1.length;
+ return reusableResultTuple;
}
@Override
- public Tuple2<byte[], Integer> takeResult() throws Exception {
- byte[] result = resultBuffer.take();
- this.reusableResultTuple.f0 = result;
- this.reusableResultTuple.f1 = result.length;
- return this.reusableResultTuple;
+ public Tuple3<String, byte[], Integer> takeResult() throws Exception {
+ Tuple2<String, byte[]> result = resultBuffer.take();
+ reusableResultTuple.f0 = result.f0;
+ reusableResultTuple.f1 = result.f1;
+ reusableResultTuple.f2 = result.f1.length;
+ return reusableResultTuple;
}
@Override
@@ -372,7 +378,7 @@ public abstract class BeamPythonFunctionRunner implements PythonFunctionRunner {
/** Interrupts the progress of takeResult. */
public void notifyNoMoreResults() {
- resultBuffer.add(new byte[0]);
+ resultBuffer.add(Tuple2.of(null, new byte[0]));
}
private void finishBundle() {
@@ -444,6 +450,19 @@ public abstract class BeamPythonFunctionRunner implements PythonFunctionRunner {
.putCoders(OUTPUT_CODER_ID, createCoderProto(outputCoderDescriptor))
.putCoders(WINDOW_CODER_ID, getWindowCoderProto());
+ for (Map.Entry<String, FlinkFnApi.CoderInfoDescriptor> entry :
+ sideOutputCoderDescriptors.entrySet()) {
+ String collectionId = entry.getKey();
+ String collectionCoderId = SIDE_OUTPUT_CODER_PREFIX + collectionId;
+ componentsBuilder.putPcollections(
+ collectionId,
+ RunnerApi.PCollection.newBuilder()
+ .setWindowingStrategyId(WINDOW_STRATEGY)
+ .setCoderId(collectionCoderId)
+ .build());
+ componentsBuilder.putCoders(collectionCoderId, createCoderProto(entry.getValue()));
+ }
+
getOptionalTimerCoderProto()
.ifPresent(
timerCoderProto -> {
@@ -475,11 +494,18 @@ public abstract class BeamPythonFunctionRunner implements PythonFunctionRunner {
components.getTransformsMap().keySet().stream()
.map(id -> PipelineNode.pTransform(id, components.getTransformsOrThrow(id)))
.collect(Collectors.toList());
- List<PipelineNode.PCollectionNode> outputs =
- Collections.singletonList(
- PipelineNode.pCollection(
- OUTPUT_COLLECTION_ID,
- components.getPcollectionsOrThrow(OUTPUT_COLLECTION_ID)));
+ List<PipelineNode.PCollectionNode> outputs = new ArrayList<>();
+ outputs.add(
+ PipelineNode.pCollection(
+ OUTPUT_COLLECTION_ID,
+ components.getPcollectionsOrThrow(OUTPUT_COLLECTION_ID)));
+ for (Map.Entry<String, FlinkFnApi.CoderInfoDescriptor> entry :
+ sideOutputCoderDescriptors.entrySet()) {
+ String collectionId = entry.getKey();
+ outputs.add(
+ PipelineNode.pCollection(
+ collectionId, components.getPcollectionsOrThrow(collectionId)));
+ }
return ImmutableExecutableStage.of(
components,
environment,
@@ -501,14 +527,16 @@ public abstract class BeamPythonFunctionRunner implements PythonFunctionRunner {
ByteArrayOutputStream baos = new ByteArrayOutputStream();
windowedValueCoder.encode(value, baos);
- return Arrays.asList(
+ List<RunnerApi.ExecutableStagePayload.WireCoderSetting> settings = new ArrayList<>();
+ settings.add(
RunnerApi.ExecutableStagePayload.WireCoderSetting.newBuilder()
.setUrn(getUrn(RunnerApi.StandardCoders.Enum.PARAM_WINDOWED_VALUE))
.setPayload(
org.apache.beam.vendor.grpc.v1p26p0.com.google.protobuf.ByteString
.copyFrom(baos.toByteArray()))
.setInputOrOutputId(INPUT_COLLECTION_ID)
- .build(),
+ .build());
+ settings.add(
RunnerApi.ExecutableStagePayload.WireCoderSetting.newBuilder()
.setUrn(getUrn(RunnerApi.StandardCoders.Enum.PARAM_WINDOWED_VALUE))
.setPayload(
@@ -516,6 +544,18 @@ public abstract class BeamPythonFunctionRunner implements PythonFunctionRunner {
.copyFrom(baos.toByteArray()))
.setInputOrOutputId(OUTPUT_COLLECTION_ID)
.build());
+ for (Map.Entry<String, FlinkFnApi.CoderInfoDescriptor> entry :
+ sideOutputCoderDescriptors.entrySet()) {
+ settings.add(
+ RunnerApi.ExecutableStagePayload.WireCoderSetting.newBuilder()
+ .setUrn(getUrn(RunnerApi.StandardCoders.Enum.PARAM_WINDOWED_VALUE))
+ .setPayload(
+ org.apache.beam.vendor.grpc.v1p26p0.com.google.protobuf
+ .ByteString.copyFrom(baos.toByteArray()))
+ .setInputOrOutputId(entry.getKey())
+ .build());
+ }
+ return settings;
}
/** Gets the proto representation of the window coder. */
@@ -546,7 +586,7 @@ public abstract class BeamPythonFunctionRunner implements PythonFunctionRunner {
@Override
public FnDataReceiver<WindowedValue<byte[]>> create(String pCollectionId) {
return input -> {
- resultBuffer.add(input.getValue());
+ resultBuffer.add(Tuple2.of(pCollectionId, input.getValue()));
};
}
};
diff --git a/flink-python/src/main/java/org/apache/flink/streaming/api/utils/ProtoUtils.java b/flink-python/src/main/java/org/apache/flink/streaming/api/utils/ProtoUtils.java
index e5e482a6aa1..0f63ebb8f2a 100644
--- a/flink-python/src/main/java/org/apache/flink/streaming/api/utils/ProtoUtils.java
+++ b/flink-python/src/main/java/org/apache/flink/streaming/api/utils/ProtoUtils.java
@@ -135,6 +135,7 @@ public enum ProtoUtils {
boolean inBatchExecutionMode,
boolean isMetricEnabled,
boolean isProfileEnabled,
+ boolean hasSideOutput,
int stateCacheSize,
int mapStateReadCacheSize,
int mapStateWriteCacheSize) {
@@ -182,6 +183,7 @@ public enum ProtoUtils {
.getSerializedPythonFunction()));
builder.setMetricEnabled(isMetricEnabled);
builder.setProfileEnabled(isProfileEnabled);
+ builder.setHasSideOutput(hasSideOutput);
builder.setStateCacheSize(stateCacheSize);
builder.setMapStateReadCacheSize(mapStateReadCacheSize);
builder.setMapStateWriteCacheSize(mapStateWriteCacheSize);
@@ -204,6 +206,7 @@ public enum ProtoUtils {
boolean inBatchExecutionMode,
boolean isMetricEnabled,
boolean isProfileEnabled,
+ boolean hasSideOutput,
int stateCacheSize,
int mapStateReadCacheSize,
int mapStateWriteCacheSize) {
@@ -220,6 +223,7 @@ public enum ProtoUtils {
inBatchExecutionMode,
isMetricEnabled,
isProfileEnabled,
+ false,
stateCacheSize,
mapStateReadCacheSize,
mapStateWriteCacheSize));
@@ -233,6 +237,7 @@ public enum ProtoUtils {
inBatchExecutionMode,
isMetricEnabled,
isProfileEnabled,
+ hasSideOutput,
stateCacheSize,
mapStateReadCacheSize,
mapStateWriteCacheSize));
@@ -248,6 +253,7 @@ public enum ProtoUtils {
boolean inBatchExecutionMode,
boolean isMetricEnabled,
boolean isProfileEnabled,
+ boolean hasSideOutput,
int stateCacheSize,
int mapStateReadCacheSize,
int mapStateWriteCacheSize) {
@@ -259,6 +265,7 @@ public enum ProtoUtils {
inBatchExecutionMode,
isMetricEnabled,
isProfileEnabled,
+ hasSideOutput,
stateCacheSize,
mapStateReadCacheSize,
mapStateWriteCacheSize);
diff --git a/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/AbstractPythonStreamGroupAggregateOperator.java b/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/AbstractPythonStreamGroupAggregateOperator.java
index 5ff05306424..c72f2d51836 100644
--- a/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/AbstractPythonStreamGroupAggregateOperator.java
+++ b/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/AbstractPythonStreamGroupAggregateOperator.java
@@ -22,7 +22,7 @@ import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
-import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.fnexecution.v1.FlinkFnApi;
import org.apache.flink.runtime.state.VoidNamespace;
@@ -146,9 +146,9 @@ public abstract class AbstractPythonStreamGroupAggregateOperator
}
@Override
- public void emitResult(Tuple2<byte[], Integer> resultTuple) throws Exception {
- byte[] rawUdfResult = resultTuple.f0;
- int length = resultTuple.f1;
+ public void emitResult(Tuple3<String, byte[], Integer> resultTuple) throws Exception {
+ byte[] rawUdfResult = resultTuple.f1;
+ int length = resultTuple.f2;
bais.setBuffer(rawUdfResult, 0, length);
RowData udfResult = udfOutputTypeSerializer.deserialize(baisWrapper);
rowDataWrapper.collect(udfResult);
diff --git a/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/PythonStreamGroupWindowAggregateOperator.java b/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/PythonStreamGroupWindowAggregateOperator.java
index 5e8a0c2e2d8..48532cbbc60 100644
--- a/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/PythonStreamGroupWindowAggregateOperator.java
+++ b/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/PythonStreamGroupWindowAggregateOperator.java
@@ -22,7 +22,7 @@ import org.apache.flink.annotation.Internal;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.fnexecution.v1.FlinkFnApi;
import org.apache.flink.streaming.api.operators.InternalTimer;
@@ -363,9 +363,9 @@ public class PythonStreamGroupWindowAggregateOperator<K, W extends Window>
}
@Override
- public void emitResult(Tuple2<byte[], Integer> resultTuple) throws Exception {
- byte[] rawUdfResult = resultTuple.f0;
- int length = resultTuple.f1;
+ public void emitResult(Tuple3<String, byte[], Integer> resultTuple) throws Exception {
+ byte[] rawUdfResult = resultTuple.f1;
+ int length = resultTuple.f2;
bais.setBuffer(rawUdfResult, 0, length);
RowData udfResult = udfOutputTypeSerializer.deserialize(baisWrapper);
byte recordType = udfResult.getByte(0);
diff --git a/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/arrow/batch/BatchArrowPythonGroupAggregateFunctionOperator.java b/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/arrow/batch/BatchArrowPythonGroupAggregateFunctionOperator.java
index d38372affc0..61cb4c8a03c 100644
--- a/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/arrow/batch/BatchArrowPythonGroupAggregateFunctionOperator.java
+++ b/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/arrow/batch/BatchArrowPythonGroupAggregateFunctionOperator.java
@@ -19,7 +19,7 @@
package org.apache.flink.table.runtime.operators.python.aggregate.arrow.batch;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.binary.BinaryRowData;
@@ -89,9 +89,9 @@ public class BatchArrowPythonGroupAggregateFunctionOperator
@Override
@SuppressWarnings("ConstantConditions")
- public void emitResult(Tuple2<byte[], Integer> resultTuple) throws Exception {
- byte[] udafResult = resultTuple.f0;
- int length = resultTuple.f1;
+ public void emitResult(Tuple3<String, byte[], Integer> resultTuple) throws Exception {
+ byte[] udafResult = resultTuple.f1;
+ int length = resultTuple.f2;
bais.setBuffer(udafResult, 0, length);
int rowCount = arrowSerializer.load();
for (int i = 0; i < rowCount; i++) {
diff --git a/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/arrow/batch/BatchArrowPythonGroupWindowAggregateFunctionOperator.java b/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/arrow/batch/BatchArrowPythonGroupWindowAggregateFunctionOperator.java
index 75218bd3819..301fd476cad 100644
--- a/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/arrow/batch/BatchArrowPythonGroupWindowAggregateFunctionOperator.java
+++ b/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/arrow/batch/BatchArrowPythonGroupWindowAggregateFunctionOperator.java
@@ -20,6 +20,7 @@ package org.apache.flink.table.runtime.operators.python.aggregate.arrow.batch;
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
@@ -158,9 +159,9 @@ public class BatchArrowPythonGroupWindowAggregateFunctionOperator
@Override
@SuppressWarnings("ConstantConditions")
- public void emitResult(Tuple2<byte[], Integer> resultTuple) throws Exception {
- byte[] udafResult = resultTuple.f0;
- int length = resultTuple.f1;
+ public void emitResult(Tuple3<String, byte[], Integer> resultTuple) throws Exception {
+ byte[] udafResult = resultTuple.f1;
+ int length = resultTuple.f2;
bais.setBuffer(udafResult, 0, length);
int rowCount = arrowSerializer.load();
for (int i = 0; i < rowCount; i++) {
diff --git a/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/arrow/batch/BatchArrowPythonOverWindowAggregateFunctionOperator.java b/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/arrow/batch/BatchArrowPythonOverWindowAggregateFunctionOperator.java
index 1425a35fe95..587b20a291a 100644
--- a/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/arrow/batch/BatchArrowPythonOverWindowAggregateFunctionOperator.java
+++ b/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/arrow/batch/BatchArrowPythonOverWindowAggregateFunctionOperator.java
@@ -20,7 +20,7 @@ package org.apache.flink.table.runtime.operators.python.aggregate.arrow.batch;
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.typeutils.base.IntSerializer;
-import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.memory.ByteArrayOutputStreamWithPos;
import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
@@ -236,9 +236,9 @@ public class BatchArrowPythonOverWindowAggregateFunctionOperator
@Override
@SuppressWarnings("ConstantConditions")
- public void emitResult(Tuple2<byte[], Integer> resultTuple) throws Exception {
- byte[] udafResult = resultTuple.f0;
- int length = resultTuple.f1;
+ public void emitResult(Tuple3<String, byte[], Integer> resultTuple) throws Exception {
+ byte[] udafResult = resultTuple.f1;
+ int length = resultTuple.f2;
bais.setBuffer(udafResult, 0, length);
int rowCount = arrowSerializer.load();
for (int i = 0; i < rowCount; i++) {
diff --git a/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/arrow/stream/AbstractStreamArrowPythonBoundedRangeOperator.java b/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/arrow/stream/AbstractStreamArrowPythonBoundedRangeOperator.java
index 67d6748cbe1..e7be013c712 100644
--- a/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/arrow/stream/AbstractStreamArrowPythonBoundedRangeOperator.java
+++ b/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/arrow/stream/AbstractStreamArrowPythonBoundedRangeOperator.java
@@ -19,7 +19,7 @@
package org.apache.flink.table.runtime.operators.python.aggregate.arrow.stream;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.VoidNamespace;
import org.apache.flink.streaming.api.TimeDomain;
@@ -113,9 +113,9 @@ public abstract class AbstractStreamArrowPythonBoundedRangeOperator<K>
@Override
@SuppressWarnings("ConstantConditions")
- public void emitResult(Tuple2<byte[], Integer> resultTuple) throws Exception {
- byte[] udafResult = resultTuple.f0;
- int length = resultTuple.f1;
+ public void emitResult(Tuple3<String, byte[], Integer> resultTuple) throws Exception {
+ byte[] udafResult = resultTuple.f1;
+ int length = resultTuple.f2;
bais.setBuffer(udafResult, 0, length);
int rowCount = arrowSerializer.load();
for (int i = 0; i < rowCount; i++) {
diff --git a/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/arrow/stream/AbstractStreamArrowPythonBoundedRowsOperator.java b/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/arrow/stream/AbstractStreamArrowPythonBoundedRowsOperator.java
index 1c8dc27e56a..a1f10f3d7a3 100644
--- a/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/arrow/stream/AbstractStreamArrowPythonBoundedRowsOperator.java
+++ b/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/arrow/stream/AbstractStreamArrowPythonBoundedRowsOperator.java
@@ -19,7 +19,7 @@
package org.apache.flink.table.runtime.operators.python.aggregate.arrow.stream;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.VoidNamespace;
import org.apache.flink.streaming.api.operators.InternalTimer;
@@ -144,9 +144,9 @@ public abstract class AbstractStreamArrowPythonBoundedRowsOperator<K>
@Override
@SuppressWarnings("ConstantConditions")
- public void emitResult(Tuple2<byte[], Integer> resultTuple) throws Exception {
- byte[] udafResult = resultTuple.f0;
- int length = resultTuple.f1;
+ public void emitResult(Tuple3<String, byte[], Integer> resultTuple) throws Exception {
+ byte[] udafResult = resultTuple.f1;
+ int length = resultTuple.f2;
bais.setBuffer(udafResult, 0, length);
int rowCount = arrowSerializer.load();
for (int i = 0; i < rowCount; i++) {
diff --git a/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/arrow/stream/StreamArrowPythonGroupWindowAggregateFunctionOperator.java b/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/arrow/stream/StreamArrowPythonGroupWindowAggregateFunctionOperator.java
index 99407ed33be..cad685493e3 100644
--- a/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/arrow/stream/StreamArrowPythonGroupWindowAggregateFunctionOperator.java
+++ b/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/arrow/stream/StreamArrowPythonGroupWindowAggregateFunctionOperator.java
@@ -26,6 +26,7 @@ import org.apache.flink.api.common.state.State;
import org.apache.flink.api.common.state.StateDescriptor;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.runtime.state.internal.InternalListState;
@@ -240,9 +241,9 @@ public class StreamArrowPythonGroupWindowAggregateFunctionOperator<K, W extends
@Override
@SuppressWarnings("ConstantConditions")
- public void emitResult(Tuple2<byte[], Integer> resultTuple) throws Exception {
- byte[] udafResult = resultTuple.f0;
- int length = resultTuple.f1;
+ public void emitResult(Tuple3<String, byte[], Integer> resultTuple) throws Exception {
+ byte[] udafResult = resultTuple.f1;
+ int length = resultTuple.f2;
bais.setBuffer(udafResult, 0, length);
int rowCount = arrowSerializer.load();
for (int i = 0; i < rowCount; i++) {
diff --git a/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/scalar/PythonScalarFunctionOperator.java b/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/scalar/PythonScalarFunctionOperator.java
index 0d72dd0d625..6c412d98363 100644
--- a/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/scalar/PythonScalarFunctionOperator.java
+++ b/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/scalar/PythonScalarFunctionOperator.java
@@ -20,7 +20,7 @@ package org.apache.flink.table.runtime.operators.python.scalar;
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.fnexecution.v1.FlinkFnApi;
import org.apache.flink.table.data.RowData;
@@ -100,9 +100,9 @@ public class PythonScalarFunctionOperator extends AbstractPythonScalarFunctionOp
@Override
@SuppressWarnings("ConstantConditions")
- public void emitResult(Tuple2<byte[], Integer> resultTuple) throws IOException {
- byte[] rawUdfResult = resultTuple.f0;
- int length = resultTuple.f1;
+ public void emitResult(Tuple3<String, byte[], Integer> resultTuple) throws IOException {
+ byte[] rawUdfResult = resultTuple.f1;
+ int length = resultTuple.f2;
RowData input = forwardedInputQueue.poll();
reuseJoinedRow.setRowKind(input.getRowKind());
bais.setBuffer(rawUdfResult, 0, length);
diff --git a/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/scalar/arrow/ArrowPythonScalarFunctionOperator.java b/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/scalar/arrow/ArrowPythonScalarFunctionOperator.java
index 37e9010f227..bc8c67bb2ec 100644
--- a/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/scalar/arrow/ArrowPythonScalarFunctionOperator.java
+++ b/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/scalar/arrow/ArrowPythonScalarFunctionOperator.java
@@ -19,7 +19,7 @@
package org.apache.flink.table.runtime.operators.python.scalar.arrow;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.fnexecution.v1.FlinkFnApi;
import org.apache.flink.table.data.RowData;
@@ -116,9 +116,9 @@ public class ArrowPythonScalarFunctionOperator extends AbstractPythonScalarFunct
@Override
@SuppressWarnings("ConstantConditions")
- public void emitResult(Tuple2<byte[], Integer> resultTuple) throws Exception {
- byte[] udfResult = resultTuple.f0;
- int length = resultTuple.f1;
+ public void emitResult(Tuple3<String, byte[], Integer> resultTuple) throws Exception {
+ byte[] udfResult = resultTuple.f1;
+ int length = resultTuple.f2;
bais.setBuffer(udfResult, 0, length);
int rowCount = arrowSerializer.load();
for (int i = 0; i < rowCount; i++) {
diff --git a/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/table/PythonTableFunctionOperator.java b/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/table/PythonTableFunctionOperator.java
index 81862b99300..732674da893 100644
--- a/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/table/PythonTableFunctionOperator.java
+++ b/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/table/PythonTableFunctionOperator.java
@@ -20,7 +20,7 @@ package org.apache.flink.table.runtime.operators.python.table;
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.fnexecution.v1.FlinkFnApi;
import org.apache.flink.streaming.api.utils.ProtoUtils;
@@ -186,7 +186,7 @@ public class PythonTableFunctionOperator
@Override
@SuppressWarnings("ConstantConditions")
- public void emitResult(Tuple2<byte[], Integer> resultTuple) throws Exception {
+ public void emitResult(Tuple3<String, byte[], Integer> resultTuple) throws Exception {
byte[] rawUdtfResult;
int length;
if (isFinishResult) {
@@ -194,8 +194,8 @@ public class PythonTableFunctionOperator
hasJoined = false;
}
do {
- rawUdtfResult = resultTuple.f0;
- length = resultTuple.f1;
+ rawUdtfResult = resultTuple.f1;
+ length = resultTuple.f2;
isFinishResult = isFinishResult(rawUdtfResult, length);
if (!isFinishResult) {
reuseJoinedRow.setRowKind(input.getRowKind());
diff --git a/flink-python/src/main/java/org/apache/flink/table/runtime/runners/python/beam/BeamTablePythonFunctionRunner.java b/flink-python/src/main/java/org/apache/flink/table/runtime/runners/python/beam/BeamTablePythonFunctionRunner.java
index ff605356d34..a1e6ff26542 100644
--- a/flink-python/src/main/java/org/apache/flink/table/runtime/runners/python/beam/BeamTablePythonFunctionRunner.java
+++ b/flink-python/src/main/java/org/apache/flink/table/runtime/runners/python/beam/BeamTablePythonFunctionRunner.java
@@ -75,7 +75,8 @@ public class BeamTablePythonFunctionRunner extends BeamPythonFunctionRunner {
memoryManager,
managedMemoryFraction,
inputCoderDescriptor,
- outputCoderDescriptor);
+ outputCoderDescriptor,
+ Collections.emptyMap());
this.functionUrn = Preconditions.checkNotNull(functionUrn);
this.userDefinedFunctionProto = Preconditions.checkNotNull(userDefinedFunctionProto);
}
diff --git a/flink-python/src/test/java/org/apache/flink/python/util/DataStreamTestCollectSink.java b/flink-python/src/test/java/org/apache/flink/python/util/DataStreamTestCollectSink.java
index dee24e462d4..a692f59b918 100644
--- a/flink-python/src/test/java/org/apache/flink/python/util/DataStreamTestCollectSink.java
+++ b/flink-python/src/test/java/org/apache/flink/python/util/DataStreamTestCollectSink.java
@@ -17,15 +17,23 @@
package org.apache.flink.python.util;
+import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import java.util.ArrayList;
import java.util.List;
+import java.util.stream.Collectors;
/** A SinkFunction for collecting results of DataStream transformations in test cases. */
public class DataStreamTestCollectSink<IN> implements SinkFunction<IN> {
- private static List<Object> collectedResult = new ArrayList<>();
+ private final int id;
+
+ private static final List<Tuple2<Integer, Object>> collectedResult = new ArrayList<>();
+
+ public DataStreamTestCollectSink() {
+ id = hashCode();
+ }
/**
* Collect the sink value into a static List so that the client side can fetch the result of
@@ -37,17 +45,22 @@ public class DataStreamTestCollectSink<IN> implements SinkFunction<IN> {
public void invoke(IN value, Context context) throws Exception {
synchronized (collectedResult) {
- collectedResult.add(value);
+ collectedResult.add(Tuple2.of(id, value));
}
}
public List<Object> collectAndClear(boolean isPythonObjects) {
- List<Object> listToBeReturned = new ArrayList<>();
- if (isPythonObjects) {
- listToBeReturned.addAll(collectedResult);
- } else {
- for (Object obj : collectedResult) {
- listToBeReturned.add(obj.toString());
+ List<Object> listToBeReturned;
+ synchronized (collectedResult) {
+ listToBeReturned =
+ collectedResult.stream()
+ .filter(e -> e.f0.equals(id))
+ .map(e -> e.f1)
+ .collect(Collectors.toList());
+ }
+ if (!isPythonObjects) {
+ for (int i = 0; i < listToBeReturned.size(); i++) {
+ listToBeReturned.set(i, listToBeReturned.get(i).toString());
}
}
clear();
@@ -55,6 +68,8 @@ public class DataStreamTestCollectSink<IN> implements SinkFunction<IN> {
}
public void clear() {
- collectedResult.clear();
+ synchronized (collectedResult) {
+ collectedResult.removeIf(e -> e.f0.equals(id));
+ }
}
}
diff --git a/flink-python/src/test/java/org/apache/flink/streaming/api/functions/python/eventtime/PerElementWatermarkGenerator.java b/flink-python/src/test/java/org/apache/flink/streaming/api/functions/python/eventtime/PerElementWatermarkGenerator.java
new file mode 100644
index 00000000000..b3b7a130ccc
--- /dev/null
+++ b/flink-python/src/test/java/org/apache/flink/streaming/api/functions/python/eventtime/PerElementWatermarkGenerator.java
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.functions.python.eventtime;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.eventtime.Watermark;
+import org.apache.flink.api.common.eventtime.WatermarkGenerator;
+import org.apache.flink.api.common.eventtime.WatermarkGeneratorSupplier;
+import org.apache.flink.api.common.eventtime.WatermarkOutput;
+
+/** PerElementWatermarkGenerator that generates max watermark for every arriving item. */
+@Internal
+public class PerElementWatermarkGenerator implements WatermarkGenerator<Object> {
+
+ private long maxTimestamp;
+
+ public PerElementWatermarkGenerator() {
+ maxTimestamp = Long.MIN_VALUE;
+ }
+
+ @Override
+ public void onEvent(Object event, long eventTimestamp, WatermarkOutput output) {
+ if (eventTimestamp > maxTimestamp) {
+ maxTimestamp = eventTimestamp;
+ output.emitWatermark(new Watermark(eventTimestamp));
+ }
+ }
+
+ @Override
+ public void onPeriodicEmit(WatermarkOutput output) {}
+
+ public static WatermarkGeneratorSupplier<Object> getSupplier() {
+ return (ctx) -> new PerElementWatermarkGenerator();
+ }
+}
diff --git a/flink-python/src/test/java/org/apache/flink/table/runtime/operators/python/aggregate/PassThroughPythonStreamGroupWindowAggregateOperator.java b/flink-python/src/test/java/org/apache/flink/table/runtime/operators/python/aggregate/PassThroughPythonStreamGroupWindowAggregateOperator.java
index a14e36375e8..cad41063250 100644
--- a/flink-python/src/test/java/org/apache/flink/table/runtime/operators/python/aggregate/PassThroughPythonStreamGroupWindowAggregateOperator.java
+++ b/flink-python/src/test/java/org/apache/flink/table/runtime/operators/python/aggregate/PassThroughPythonStreamGroupWindowAggregateOperator.java
@@ -18,6 +18,7 @@
package org.apache.flink.table.runtime.operators.python.aggregate;
+import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.memory.ByteArrayOutputStreamWithPos;
import org.apache.flink.core.memory.DataInputDeserializer;
@@ -68,6 +69,7 @@ import java.util.concurrent.LinkedBlockingQueue;
import java.util.function.Function;
import java.util.stream.Collectors;
+import static org.apache.flink.python.Constants.OUTPUT_COLLECTION_ID;
import static org.apache.flink.table.runtime.util.TimeWindowUtil.toEpochMillsForTimer;
/** PassThroughPythonStreamGroupWindowAggregateOperator. */
@@ -85,7 +87,7 @@ public class PassThroughPythonStreamGroupWindowAggregateOperator<K>
private transient UpdatableRowData reusePythonTimerRowData;
private transient UpdatableRowData reusePythonTimerData;
- private transient LinkedBlockingQueue<byte[]> resultBuffer;
+ private transient LinkedBlockingQueue<Tuple2<String, byte[]>> resultBuffer;
private Projection<RowData, BinaryRowData> groupKeyProjection;
private Function<RowData, RowData> aggExtracter;
private Function<TimeWindow, RowData> windowExtractor;
@@ -294,7 +296,7 @@ public class PassThroughPythonStreamGroupWindowAggregateOperator<K>
}
}
- public void setResultBuffer(LinkedBlockingQueue<byte[]> resultBuffer) {
+ public void setResultBuffer(LinkedBlockingQueue<Tuple2<String, byte[]>> resultBuffer) {
this.resultBuffer = resultBuffer;
}
@@ -342,7 +344,7 @@ public class PassThroughPythonStreamGroupWindowAggregateOperator<K>
reuseJoinedRow.replace(windowAggResult, windowProperty);
reusePythonRowData.setField(1, reuseJoinedRow);
udfOutputTypeSerializer.serialize(reusePythonRowData, output);
- resultBuffer.add(output.getCopyOfBuffer());
+ resultBuffer.add(Tuple2.of(OUTPUT_COLLECTION_ID, output.getCopyOfBuffer()));
break;
}
}
@@ -425,7 +427,7 @@ public class PassThroughPythonStreamGroupWindowAggregateOperator<K>
windowBaos.reset();
DataOutputSerializer output = new DataOutputSerializer(1);
udfOutputTypeSerializer.serialize(reusePythonTimerRowData, output);
- resultBuffer.add(output.getCopyOfBuffer());
+ resultBuffer.add(Tuple2.of(OUTPUT_COLLECTION_ID, output.getCopyOfBuffer()));
}
private void cleanWindowIfNeeded(RowData key, TimeWindow window, long currentTime)
@@ -446,7 +448,7 @@ public class PassThroughPythonStreamGroupWindowAggregateOperator<K>
reuseJoinedRow.replace(windowAggResult, windowProperty);
reusePythonRowData.setField(1, reuseJoinedRow);
udfOutputTypeSerializer.serialize(reusePythonRowData, output);
- resultBuffer.add(output.getCopyOfBuffer());
+ resultBuffer.add(Tuple2.of(OUTPUT_COLLECTION_ID, output.getCopyOfBuffer()));
// 2. delete window timer
if (windowAssigner.isEventTime()) {
deleteEventTimeTimer(key, window);
diff --git a/flink-python/src/test/java/org/apache/flink/table/runtime/utils/PassThroughPythonAggregateFunctionRunner.java b/flink-python/src/test/java/org/apache/flink/table/runtime/utils/PassThroughPythonAggregateFunctionRunner.java
index f19b5c60188..a1667c9c9c3 100644
--- a/flink-python/src/test/java/org/apache/flink/table/runtime/utils/PassThroughPythonAggregateFunctionRunner.java
+++ b/flink-python/src/test/java/org/apache/flink/table/runtime/utils/PassThroughPythonAggregateFunctionRunner.java
@@ -19,6 +19,7 @@
package org.apache.flink.table.runtime.utils;
import org.apache.flink.api.common.typeutils.base.IntSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.core.memory.ByteArrayInputStreamWithPos;
import org.apache.flink.core.memory.ByteArrayOutputStreamWithPos;
@@ -37,7 +38,9 @@ import org.apache.beam.vendor.grpc.v1p26p0.com.google.protobuf.Struct;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
+import java.util.stream.Collectors;
+import static org.apache.flink.python.Constants.OUTPUT_COLLECTION_ID;
import static org.apache.flink.streaming.api.utils.ProtoUtils.createArrowTypeCoderInfoDescriptorProto;
/**
@@ -144,7 +147,10 @@ public class PassThroughPythonAggregateFunctionRunner extends BeamTablePythonFun
@Override
public void flush() throws Exception {
super.flush();
- resultBuffer.addAll(buffer);
+ resultBuffer.addAll(
+ buffer.stream()
+ .map(b -> Tuple2.of(OUTPUT_COLLECTION_ID, b))
+ .collect(Collectors.toList()));
buffer.clear();
}
diff --git a/flink-python/src/test/java/org/apache/flink/table/runtime/utils/PassThroughPythonScalarFunctionRunner.java b/flink-python/src/test/java/org/apache/flink/table/runtime/utils/PassThroughPythonScalarFunctionRunner.java
index e0cd0fbde1c..f1f6bb6710d 100644
--- a/flink-python/src/test/java/org/apache/flink/table/runtime/utils/PassThroughPythonScalarFunctionRunner.java
+++ b/flink-python/src/test/java/org/apache/flink/table/runtime/utils/PassThroughPythonScalarFunctionRunner.java
@@ -18,6 +18,7 @@
package org.apache.flink.table.runtime.utils;
+import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.fnexecution.v1.FlinkFnApi;
import org.apache.flink.python.env.process.ProcessPythonEnvironmentManager;
import org.apache.flink.python.metric.FlinkMetricContainer;
@@ -29,7 +30,9 @@ import org.apache.beam.vendor.grpc.v1p26p0.com.google.protobuf.Struct;
import java.util.LinkedList;
import java.util.List;
+import java.util.stream.Collectors;
+import static org.apache.flink.python.Constants.OUTPUT_COLLECTION_ID;
import static org.apache.flink.streaming.api.utils.ProtoUtils.createFlattenRowTypeCoderInfoDescriptorProto;
/**
@@ -75,7 +78,10 @@ public class PassThroughPythonScalarFunctionRunner extends BeamTablePythonFuncti
@Override
public void flush() throws Exception {
super.flush();
- resultBuffer.addAll(buffer);
+ resultBuffer.addAll(
+ buffer.stream()
+ .map(b -> Tuple2.of(OUTPUT_COLLECTION_ID, b))
+ .collect(Collectors.toList()));
buffer.clear();
}
diff --git a/flink-python/src/test/java/org/apache/flink/table/runtime/utils/PassThroughPythonTableFunctionRunner.java b/flink-python/src/test/java/org/apache/flink/table/runtime/utils/PassThroughPythonTableFunctionRunner.java
index 48af1207b6e..53713d8b72b 100644
--- a/flink-python/src/test/java/org/apache/flink/table/runtime/utils/PassThroughPythonTableFunctionRunner.java
+++ b/flink-python/src/test/java/org/apache/flink/table/runtime/utils/PassThroughPythonTableFunctionRunner.java
@@ -18,6 +18,7 @@
package org.apache.flink.table.runtime.utils;
+import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.fnexecution.v1.FlinkFnApi;
import org.apache.flink.python.env.process.ProcessPythonEnvironmentManager;
import org.apache.flink.python.metric.FlinkMetricContainer;
@@ -29,7 +30,9 @@ import org.apache.beam.vendor.grpc.v1p26p0.com.google.protobuf.Struct;
import java.util.LinkedList;
import java.util.List;
+import java.util.stream.Collectors;
+import static org.apache.flink.python.Constants.OUTPUT_COLLECTION_ID;
import static org.apache.flink.streaming.api.utils.ProtoUtils.createFlattenRowTypeCoderInfoDescriptorProto;
/**
@@ -84,7 +87,10 @@ public class PassThroughPythonTableFunctionRunner extends BeamTablePythonFunctio
@Override
public void flush() throws Exception {
super.flush();
- resultBuffer.addAll(buffer);
+ resultBuffer.addAll(
+ buffer.stream()
+ .map(b -> Tuple2.of(OUTPUT_COLLECTION_ID, b))
+ .collect(Collectors.toList()));
buffer.clear();
}
diff --git a/flink-python/src/test/java/org/apache/flink/table/runtime/utils/PassThroughStreamAggregatePythonFunctionRunner.java b/flink-python/src/test/java/org/apache/flink/table/runtime/utils/PassThroughStreamAggregatePythonFunctionRunner.java
index 230f2e4b862..9698264a58f 100644
--- a/flink-python/src/test/java/org/apache/flink/table/runtime/utils/PassThroughStreamAggregatePythonFunctionRunner.java
+++ b/flink-python/src/test/java/org/apache/flink/table/runtime/utils/PassThroughStreamAggregatePythonFunctionRunner.java
@@ -19,6 +19,7 @@
package org.apache.flink.table.runtime.utils;
import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.fnexecution.v1.FlinkFnApi;
import org.apache.flink.python.env.process.ProcessPythonEnvironmentManager;
import org.apache.flink.python.metric.FlinkMetricContainer;
@@ -32,7 +33,9 @@ import org.apache.beam.vendor.grpc.v1p26p0.com.google.protobuf.Struct;
import java.util.LinkedList;
import java.util.List;
import java.util.function.Function;
+import java.util.stream.Collectors;
+import static org.apache.flink.python.Constants.OUTPUT_COLLECTION_ID;
import static org.apache.flink.streaming.api.utils.ProtoUtils.createRowTypeCoderInfoDescriptorProto;
/**
@@ -85,7 +88,10 @@ public class PassThroughStreamAggregatePythonFunctionRunner extends BeamTablePyt
@Override
public void flush() throws Exception {
super.flush();
- resultBuffer.addAll(buffer);
+ resultBuffer.addAll(
+ buffer.stream()
+ .map(b -> Tuple2.of(OUTPUT_COLLECTION_ID, b))
+ .collect(Collectors.toList()));
buffer.clear();
}
diff --git a/flink-python/src/test/java/org/apache/flink/table/runtime/utils/PassThroughStreamTableAggregatePythonFunctionRunner.java b/flink-python/src/test/java/org/apache/flink/table/runtime/utils/PassThroughStreamTableAggregatePythonFunctionRunner.java
index d519a1a220f..77ea13237c5 100644
--- a/flink-python/src/test/java/org/apache/flink/table/runtime/utils/PassThroughStreamTableAggregatePythonFunctionRunner.java
+++ b/flink-python/src/test/java/org/apache/flink/table/runtime/utils/PassThroughStreamTableAggregatePythonFunctionRunner.java
@@ -19,6 +19,7 @@
package org.apache.flink.table.runtime.utils;
import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.fnexecution.v1.FlinkFnApi;
import org.apache.flink.python.env.process.ProcessPythonEnvironmentManager;
import org.apache.flink.python.metric.FlinkMetricContainer;
@@ -33,7 +34,9 @@ import java.util.Arrays;
import java.util.LinkedList;
import java.util.List;
import java.util.function.Function;
+import java.util.stream.Collectors;
+import static org.apache.flink.python.Constants.OUTPUT_COLLECTION_ID;
import static org.apache.flink.streaming.api.utils.ProtoUtils.createRowTypeCoderInfoDescriptorProto;
/**
@@ -91,7 +94,10 @@ public class PassThroughStreamTableAggregatePythonFunctionRunner
@Override
public void flush() throws Exception {
super.flush();
- resultBuffer.addAll(buffer);
+ resultBuffer.addAll(
+ buffer.stream()
+ .map(b -> Tuple2.of(OUTPUT_COLLECTION_ID, b))
+ .collect(Collectors.toList()));
buffer.clear();
}