You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by he...@apache.org on 2020/08/16 09:12:02 UTC
[flink] branch master updated (1fc400d -> a86387b)
This is an automated email from the ASF dual-hosted git repository.
hequn pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 1fc400d [FLINK-18879][python] Support Row Serialization and Deserialization schemas for Python DataStream API. (#13150)
new 175038b [FLINK-18943][python] Support CoMapFunction for Python DataStream API
new a86387b [FLINK-18943][python] Rename AbstractPythonFunctionOperator to AbstractOneInputPythonFunctionOperator
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:
flink-python/dev/glibc_version_fix.h | 0
flink-python/pyflink/datastream/data_stream.py | 105 ++++++++++++++++++++-
flink-python/pyflink/datastream/functions.py | 35 +++++++
.../pyflink/datastream/tests/test_data_stream.py | 38 ++++++++
.../pyflink/fn_execution/flink_fn_execution_pb2.py | 90 ++++++++++--------
.../pyflink/fn_execution/operation_utils.py | 6 ++
.../pyflink/proto/flink-fn-execution.proto | 2 +
.../DataStreamPythonStatelessFunctionOperator.java | 5 +-
...amTwoInputPythonStatelessFunctionOperator.java} | 87 +++++++++++------
.../AbstractOneInputPythonFunctionOperator.java | 23 ++---
...ava => AbstractPythonFunctionOperatorBase.java} | 18 +---
.../AbstractTwoInputPythonFunctionOperator.java | 26 ++---
.../python/AbstractStatelessFunctionOperator.java | 4 +-
13 files changed, 326 insertions(+), 113 deletions(-)
mode change 100644 => 100755 flink-python/dev/glibc_version_fix.h
copy flink-python/src/main/java/org/apache/flink/datastream/runtime/operators/python/{DataStreamPythonStatelessFunctionOperator.java => DataStreamTwoInputPythonStatelessFunctionOperator.java} (62%)
copy flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/data/conversion/DayTimeIntervalDurationConverter.java => flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractOneInputPythonFunctionOperator.java (58%)
rename flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/{AbstractPythonFunctionOperator.java => AbstractPythonFunctionOperatorBase.java} (95%)
copy flink-java/src/main/java/org/apache/flink/api/java/functions/FormattingMapper.java => flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractTwoInputPythonFunctionOperator.java (57%)
[flink] 02/02: [FLINK-18943][python] Rename
AbstractPythonFunctionOperator to AbstractOneInputPythonFunctionOperator
Posted by he...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hequn pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit a86387bfa9ac525223b8e16b0a747d35893c9302
Author: hequn.chq <he...@alibaba-inc.com>
AuthorDate: Sat Aug 15 12:41:57 2020 +0800
[FLINK-18943][python] Rename AbstractPythonFunctionOperator to AbstractOneInputPythonFunctionOperator
---
.../operators/python/DataStreamPythonStatelessFunctionOperator.java | 5 +++--
...tionOperator.java => AbstractOneInputPythonFunctionOperator.java} | 4 ++--
.../runtime/operators/python/AbstractStatelessFunctionOperator.java | 4 ++--
3 files changed, 7 insertions(+), 6 deletions(-)
diff --git a/flink-python/src/main/java/org/apache/flink/datastream/runtime/operators/python/DataStreamPythonStatelessFunctionOperator.java b/flink-python/src/main/java/org/apache/flink/datastream/runtime/operators/python/DataStreamPythonStatelessFunctionOperator.java
index 1b32df4..a2ec04c 100644
--- a/flink-python/src/main/java/org/apache/flink/datastream/runtime/operators/python/DataStreamPythonStatelessFunctionOperator.java
+++ b/flink-python/src/main/java/org/apache/flink/datastream/runtime/operators/python/DataStreamPythonStatelessFunctionOperator.java
@@ -30,7 +30,7 @@ import org.apache.flink.datastream.runtime.runners.python.beam.BeamDataStreamPyt
import org.apache.flink.datastream.runtime.typeutils.python.PythonTypeUtils;
import org.apache.flink.fnexecution.v1.FlinkFnApi;
import org.apache.flink.python.PythonFunctionRunner;
-import org.apache.flink.streaming.api.operators.python.AbstractPythonFunctionOperator;
+import org.apache.flink.streaming.api.operators.python.AbstractOneInputPythonFunctionOperator;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.table.functions.python.PythonEnv;
import org.apache.flink.table.runtime.util.StreamRecordCollector;
@@ -43,7 +43,8 @@ import java.util.Map;
* {@link DataStreamPythonStatelessFunctionOperator} is responsible for launching beam runner which will start a python
* harness to execute user defined python function.
*/
-public class DataStreamPythonStatelessFunctionOperator<IN, OUT> extends AbstractPythonFunctionOperator<IN, OUT> {
+public class DataStreamPythonStatelessFunctionOperator<IN, OUT>
+ extends AbstractOneInputPythonFunctionOperator<IN, OUT> {
private static final long serialVersionUID = 1L;
diff --git a/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractPythonFunctionOperator.java b/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractOneInputPythonFunctionOperator.java
similarity index 91%
rename from flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractPythonFunctionOperator.java
rename to flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractOneInputPythonFunctionOperator.java
index 1295b8f..0ae4ed9 100644
--- a/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractPythonFunctionOperator.java
+++ b/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractOneInputPythonFunctionOperator.java
@@ -27,13 +27,13 @@ import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
* Base class for all one input stream operators to execute Python functions.
*/
@Internal
-public abstract class AbstractPythonFunctionOperator<IN, OUT>
+public abstract class AbstractOneInputPythonFunctionOperator<IN, OUT>
extends AbstractPythonFunctionOperatorBase<OUT>
implements OneInputStreamOperator<IN, OUT>, BoundedOneInput {
private static final long serialVersionUID = 1L;
- public AbstractPythonFunctionOperator(Configuration config) {
+ public AbstractOneInputPythonFunctionOperator(Configuration config) {
super(config);
}
diff --git a/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/AbstractStatelessFunctionOperator.java b/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/AbstractStatelessFunctionOperator.java
index f68d8a3..a3ae165 100644
--- a/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/AbstractStatelessFunctionOperator.java
+++ b/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/AbstractStatelessFunctionOperator.java
@@ -26,7 +26,7 @@ import org.apache.flink.core.memory.DataInputViewStreamWrapper;
import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
import org.apache.flink.fnexecution.v1.FlinkFnApi;
import org.apache.flink.python.PythonFunctionRunner;
-import org.apache.flink.streaming.api.operators.python.AbstractPythonFunctionOperator;
+import org.apache.flink.streaming.api.operators.python.AbstractOneInputPythonFunctionOperator;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.functions.python.PythonFunctionInfo;
@@ -55,7 +55,7 @@ import java.util.stream.Collectors;
*/
@Internal
public abstract class AbstractStatelessFunctionOperator<IN, OUT, UDFIN>
- extends AbstractPythonFunctionOperator<IN, OUT> {
+ extends AbstractOneInputPythonFunctionOperator<IN, OUT> {
private static final long serialVersionUID = 1L;
[flink] 01/02: [FLINK-18943][python] Support CoMapFunction for
Python DataStream API
Posted by he...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hequn pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 175038b705f82655a4a082de64a73f989f6abea4
Author: hequn.chq <he...@alibaba-inc.com>
AuthorDate: Sat Aug 15 12:39:41 2020 +0800
[FLINK-18943][python] Support CoMapFunction for Python DataStream API
---
flink-python/dev/glibc_version_fix.h | 0
flink-python/pyflink/datastream/data_stream.py | 105 ++++++-
flink-python/pyflink/datastream/functions.py | 35 +++
.../pyflink/datastream/tests/test_data_stream.py | 38 +++
.../pyflink/fn_execution/flink_fn_execution_pb2.py | 90 +++---
.../pyflink/fn_execution/operation_utils.py | 6 +
.../pyflink/proto/flink-fn-execution.proto | 2 +
...eamTwoInputPythonStatelessFunctionOperator.java | 196 +++++++++++++
.../python/AbstractPythonFunctionOperator.java | 320 +--------------------
...ava => AbstractPythonFunctionOperatorBase.java} | 18 +-
.../AbstractTwoInputPythonFunctionOperator.java | 44 +++
11 files changed, 482 insertions(+), 372 deletions(-)
diff --git a/flink-python/dev/glibc_version_fix.h b/flink-python/dev/glibc_version_fix.h
old mode 100644
new mode 100755
diff --git a/flink-python/pyflink/datastream/data_stream.py b/flink-python/pyflink/datastream/data_stream.py
index 67b2903..17abb70 100644
--- a/flink-python/pyflink/datastream/data_stream.py
+++ b/flink-python/pyflink/datastream/data_stream.py
@@ -24,7 +24,7 @@ from pyflink.common.typeinfo import TypeInformation
from pyflink.datastream.functions import _get_python_env, FlatMapFunctionWrapper, FlatMapFunction, \
MapFunction, MapFunctionWrapper, Function, FunctionWrapper, SinkFunction, FilterFunction, \
FilterFunctionWrapper, KeySelectorFunctionWrapper, KeySelector, ReduceFunction, \
- ReduceFunctionWrapper
+ ReduceFunctionWrapper, CoMapFunction
from pyflink.java_gateway import get_gateway
@@ -359,6 +359,17 @@ class DataStream(object):
j_united_stream = self._j_data_stream.union(j_data_stream_arr)
return DataStream(j_data_stream=j_united_stream)
+ def connect(self, ds: 'DataStream') -> 'ConnectedStreams':
+ """
+ Creates a new 'ConnectedStreams' by connecting 'DataStream' outputs of (possible)
+ different types with each other. The DataStreams connected using this operator can
+ be used with CoFunctions to apply joint transformations.
+
+ :param ds: The DataStream with which this stream will be connected.
+ :return: The `ConnectedStreams`.
+ """
+ return ConnectedStreams(self, ds)
+
def shuffle(self) -> 'DataStream':
"""
Sets the partitioning of the DataStream so that the output elements are shuffled uniformly
@@ -687,6 +698,9 @@ class KeyedStream(DataStream):
j_python_data_stream_scalar_function_operator
))
+ def connect(self, ds: 'KeyedStream') -> 'ConnectedStreams':
+ raise Exception('Connect on KeyedStream has not been supported yet.')
+
def filter(self, func: Union[Callable, FilterFunction]) -> 'DataStream':
return self._values().filter(func)
@@ -763,3 +777,92 @@ class KeyedStream(DataStream):
def slot_sharing_group(self, slot_sharing_group: str) -> 'DataStream':
raise Exception("Setting slot sharing group for KeyedStream is not supported.")
+
+
+class ConnectedStreams(object):
+ """
+ ConnectedStreams represent two connected streams of (possibly) different data types.
+ Connected streams are useful for cases where operations on one stream directly
+ affect the operations on the other stream, usually via shared state between the streams.
+
+ An example for the use of connected streams would be to apply rules that change over time
+ onto another stream. One of the connected streams has the rules, the other stream the
+ elements to apply the rules to. The operation on the connected stream maintains the
+ current set of rules in the state. It may receive either a rule update and update the state
+ or a data element and apply the rules in the state to the element.
+
+ The connected stream can be conceptually viewed as a union stream of an Either type, that
+ holds either the first stream's type or the second stream's type.
+ """
+
+ def __init__(self, stream1: DataStream, stream2: DataStream):
+ self.stream1 = stream1
+ self.stream2 = stream2
+
+ def map(self, func: CoMapFunction, type_info: TypeInformation = None) \
+ -> 'DataStream':
+ """
+ Applies a CoMap transformation on a `ConnectedStreams` and maps the output to a common
+ type. The transformation calls a `CoMapFunction.map1` for each element of the first
+ input and `CoMapFunction.map2` for each element of the second input. Each CoMapFunction
+ call returns exactly one element.
+
+ :param func: The CoMapFunction used to jointly transform the two input DataStreams
+ :param type_info: `TypeInformation` for the result type of the function.
+ :return: The transformed `DataStream`
+ """
+ if not isinstance(func, CoMapFunction):
+ raise TypeError("The input function must be a CoMapFunction!")
+ func_name = str(func)
+
+ # get connected stream
+ j_connected_stream = self.stream1._j_data_stream.connect(self.stream2._j_data_stream)
+ from pyflink.fn_execution import flink_fn_execution_pb2
+ j_operator, j_output_type = self._get_connected_stream_operator(
+ func, type_info, func_name, flink_fn_execution_pb2.UserDefinedDataStreamFunction.CO_MAP)
+ return DataStream(j_connected_stream.transform("Co-Process", j_output_type, j_operator))
+
+ def _get_connected_stream_operator(self, func: Union[Function, FunctionWrapper],
+ type_info: TypeInformation, func_name: str,
+ func_type: int):
+ gateway = get_gateway()
+ import cloudpickle
+ serialized_func = cloudpickle.dumps(func)
+
+ j_input_types1 = self.stream1._j_data_stream.getTransformation().getOutputType()
+ j_input_types2 = self.stream2._j_data_stream.getTransformation().getOutputType()
+
+ if type_info is None:
+ output_type_info = PickledBytesTypeInfo.PICKLED_BYTE_ARRAY_TYPE_INFO()
+ else:
+ if isinstance(type_info, list):
+ output_type_info = RowTypeInfo(type_info)
+ else:
+ output_type_info = type_info
+
+ DataStreamPythonFunction = gateway.jvm.org.apache.flink.datastream.runtime.functions \
+ .python.DataStreamPythonFunction
+ j_python_data_stream_scalar_function = DataStreamPythonFunction(
+ func_name,
+ bytearray(serialized_func),
+ _get_python_env())
+
+ DataStreamPythonFunctionInfo = gateway.jvm. \
+ org.apache.flink.datastream.runtime.functions.python \
+ .DataStreamPythonFunctionInfo
+
+ j_python_data_stream_function_info = DataStreamPythonFunctionInfo(
+ j_python_data_stream_scalar_function,
+ func_type)
+
+ j_conf = gateway.jvm.org.apache.flink.configuration.Configuration()
+ DataStreamPythonFunctionOperator = gateway.jvm.org.apache.flink.datastream.runtime \
+ .operators.python.DataStreamTwoInputPythonStatelessFunctionOperator
+ j_python_data_stream_function_operator = DataStreamPythonFunctionOperator(
+ j_conf,
+ j_input_types1,
+ j_input_types2,
+ output_type_info.get_java_type_info(),
+ j_python_data_stream_function_info)
+
+ return j_python_data_stream_function_operator, output_type_info.get_java_type_info()
diff --git a/flink-python/pyflink/datastream/functions.py b/flink-python/pyflink/datastream/functions.py
index c5049e8..10433ee 100644
--- a/flink-python/pyflink/datastream/functions.py
+++ b/flink-python/pyflink/datastream/functions.py
@@ -56,6 +56,41 @@ class MapFunction(Function):
pass
+class CoMapFunction(Function):
+ """
+ A CoMapFunction implements a map() transformation over two connected streams.
+
+ The same instance of the transformation function is used to transform both of
+ the connected streams. That way, the stream transformations can share state.
+
+ The basic syntax for using a MapFunction is as follows:
+ ::
+ >>> ds1 = ...
+ >>> ds2 = ...
+ >>> new_ds = ds1.connect(ds2).map(MyCoMapFunction())
+ """
+
+ @abc.abstractmethod
+ def map1(self, value):
+ """
+ This method is called for each element in the first of the connected streams.
+
+ :param value: The stream element
+ :return: The resulting element
+ """
+ pass
+
+ @abc.abstractmethod
+ def map2(self, value):
+ """
+ This method is called for each element in the second of the connected streams.
+
+ :param value: The stream element
+ :return: The resulting element
+ """
+ pass
+
+
class FlatMapFunction(Function):
"""
Base class for flatMap functions. FLatMap functions take elements and transform them, into zero,
diff --git a/flink-python/pyflink/datastream/tests/test_data_stream.py b/flink-python/pyflink/datastream/tests/test_data_stream.py
index 937d1db..d05b2e7 100644
--- a/flink-python/pyflink/datastream/tests/test_data_stream.py
+++ b/flink-python/pyflink/datastream/tests/test_data_stream.py
@@ -22,6 +22,7 @@ from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.functions import FilterFunction
from pyflink.datastream.functions import KeySelector
from pyflink.datastream.functions import MapFunction, FlatMapFunction
+from pyflink.datastream.functions import CoMapFunction
from pyflink.datastream.tests.test_util import DataStreamTestSinkFunction
from pyflink.java_gateway import get_gateway
from pyflink.testing.test_case_utils import PyFlinkTestCase
@@ -105,6 +106,34 @@ class DataStreamTests(PyFlinkTestCase):
results.sort()
self.assertEqual(expected, results)
+ def test_co_map_function_without_data_types(self):
+ self.env.set_parallelism(1)
+ ds1 = self.env.from_collection([(1, 1), (2, 2), (3, 3)],
+ type_info=Types.ROW([Types.INT(), Types.INT()]))
+ ds2 = self.env.from_collection([("a", "a"), ("b", "b"), ("c", "c")],
+ type_info=Types.ROW([Types.STRING(), Types.STRING()]))
+ ds1.connect(ds2).map(MyCoMapFunction()).add_sink(self.test_sink)
+ self.env.execute('co_map_function_test')
+ results = self.test_sink.get_results(True)
+ expected = ['2', '3', '4', 'a', 'b', 'c']
+ expected.sort()
+ results.sort()
+ self.assertEqual(expected, results)
+
+ def test_co_map_function_with_data_types(self):
+ self.env.set_parallelism(1)
+ ds1 = self.env.from_collection([(1, 1), (2, 2), (3, 3)],
+ type_info=Types.ROW([Types.INT(), Types.INT()]))
+ ds2 = self.env.from_collection([("a", "a"), ("b", "b"), ("c", "c")],
+ type_info=Types.ROW([Types.STRING(), Types.STRING()]))
+ ds1.connect(ds2).map(MyCoMapFunction(), type_info=Types.STRING()).add_sink(self.test_sink)
+ self.env.execute('co_map_function_test')
+ results = self.test_sink.get_results(False)
+ expected = ['2', '3', '4', 'a', 'b', 'c']
+ expected.sort()
+ results.sort()
+ self.assertEqual(expected, results)
+
def test_map_function_with_data_types_and_function_object(self):
ds = self.env.from_collection([('ab', 1), ('bdc', 2), ('cfgs', 3), ('deeefg', 4)],
type_info=Types.ROW([Types.STRING(), Types.INT()]))
@@ -431,3 +460,12 @@ class MyFilterFunction(FilterFunction):
def filter(self, value):
return value[0] % 2 == 0
+
+
+class MyCoMapFunction(CoMapFunction):
+
+ def map1(self, value):
+ return str(value[0] + 1)
+
+ def map2(self, value):
+ return value[0]
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 6a7945e..f90071d 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\"\xfc\x01\n\x13UserDefinedFunction\x12\x0f\n\x07payload\x18\x01 \x01(\x0c\x12K\n\x06inputs\x18\x02 \x03(\x0b\x32;.org.apache.flink.fn_execution.v1.UserDefinedFunction.Input\x1a\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\ [...]
+ serialized_pb=_b('\n\x18\x66link-fn-execution.proto\x12 org.apache.flink.fn_execution.v1\"\xfc\x01\n\x13UserDefinedFunction\x12\x0f\n\x07payload\x18\x01 \x01(\x0c\x12K\n\x06inputs\x18\x02 \x03(\x0b\x32;.org.apache.flink.fn_execution.v1.UserDefinedFunction.Input\x1a\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\ [...]
)
@@ -59,11 +59,19 @@ _USERDEFINEDDATASTREAMFUNCTION_FUNCTIONTYPE = _descriptor.EnumDescriptor(
name='REDUCE', index=2, number=2,
options=None,
type=None),
+ _descriptor.EnumValueDescriptor(
+ name='CO_MAP', index=3, number=3,
+ options=None,
+ type=None),
+ _descriptor.EnumValueDescriptor(
+ name='CO_FLAT_MAP', index=4, number=4,
+ options=None,
+ type=None),
],
containing_type=None,
options=None,
serialized_start=585,
- serialized_end=634,
+ serialized_end=663,
)
_sym_db.RegisterEnumDescriptor(_USERDEFINEDDATASTREAMFUNCTION_FUNCTIONTYPE)
@@ -160,8 +168,8 @@ _SCHEMA_TYPENAME = _descriptor.EnumDescriptor(
],
containing_type=None,
options=None,
- serialized_start=2514,
- serialized_end=2797,
+ serialized_start=2543,
+ serialized_end=2826,
)
_sym_db.RegisterEnumDescriptor(_SCHEMA_TYPENAME)
@@ -246,8 +254,8 @@ _TYPEINFO_TYPENAME = _descriptor.EnumDescriptor(
],
containing_type=None,
options=None,
- serialized_start=3318,
- serialized_end=3549,
+ serialized_start=3347,
+ serialized_end=3578,
)
_sym_db.RegisterEnumDescriptor(_TYPEINFO_TYPENAME)
@@ -410,7 +418,7 @@ _USERDEFINEDDATASTREAMFUNCTION = _descriptor.Descriptor(
oneofs=[
],
serialized_start=435,
- serialized_end=634,
+ serialized_end=663,
)
@@ -447,8 +455,8 @@ _USERDEFINEDDATASTREAMFUNCTIONS = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=637,
- serialized_end=772,
+ serialized_start=666,
+ serialized_end=801,
)
@@ -485,8 +493,8 @@ _SCHEMA_MAPINFO = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=850,
- serialized_end=1001,
+ serialized_start=879,
+ serialized_end=1030,
)
_SCHEMA_TIMEINFO = _descriptor.Descriptor(
@@ -515,8 +523,8 @@ _SCHEMA_TIMEINFO = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=1003,
- serialized_end=1032,
+ serialized_start=1032,
+ serialized_end=1061,
)
_SCHEMA_TIMESTAMPINFO = _descriptor.Descriptor(
@@ -545,8 +553,8 @@ _SCHEMA_TIMESTAMPINFO = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=1034,
- serialized_end=1068,
+ serialized_start=1063,
+ serialized_end=1097,
)
_SCHEMA_LOCALZONEDTIMESTAMPINFO = _descriptor.Descriptor(
@@ -575,8 +583,8 @@ _SCHEMA_LOCALZONEDTIMESTAMPINFO = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=1070,
- serialized_end=1114,
+ serialized_start=1099,
+ serialized_end=1143,
)
_SCHEMA_ZONEDTIMESTAMPINFO = _descriptor.Descriptor(
@@ -605,8 +613,8 @@ _SCHEMA_ZONEDTIMESTAMPINFO = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=1116,
- serialized_end=1155,
+ serialized_start=1145,
+ serialized_end=1184,
)
_SCHEMA_DECIMALINFO = _descriptor.Descriptor(
@@ -642,8 +650,8 @@ _SCHEMA_DECIMALINFO = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=1157,
- serialized_end=1204,
+ serialized_start=1186,
+ serialized_end=1233,
)
_SCHEMA_BINARYINFO = _descriptor.Descriptor(
@@ -672,8 +680,8 @@ _SCHEMA_BINARYINFO = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=1206,
- serialized_end=1234,
+ serialized_start=1235,
+ serialized_end=1263,
)
_SCHEMA_VARBINARYINFO = _descriptor.Descriptor(
@@ -702,8 +710,8 @@ _SCHEMA_VARBINARYINFO = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=1236,
- serialized_end=1267,
+ serialized_start=1265,
+ serialized_end=1296,
)
_SCHEMA_CHARINFO = _descriptor.Descriptor(
@@ -732,8 +740,8 @@ _SCHEMA_CHARINFO = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=1269,
- serialized_end=1295,
+ serialized_start=1298,
+ serialized_end=1324,
)
_SCHEMA_VARCHARINFO = _descriptor.Descriptor(
@@ -762,8 +770,8 @@ _SCHEMA_VARCHARINFO = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=1297,
- serialized_end=1326,
+ serialized_start=1326,
+ serialized_end=1355,
)
_SCHEMA_FIELDTYPE = _descriptor.Descriptor(
@@ -886,8 +894,8 @@ _SCHEMA_FIELDTYPE = _descriptor.Descriptor(
name='type_info', full_name='org.apache.flink.fn_execution.v1.Schema.FieldType.type_info',
index=0, containing_type=None, fields=[]),
],
- serialized_start=1329,
- serialized_end=2401,
+ serialized_start=1358,
+ serialized_end=2430,
)
_SCHEMA_FIELD = _descriptor.Descriptor(
@@ -930,8 +938,8 @@ _SCHEMA_FIELD = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=2403,
- serialized_end=2511,
+ serialized_start=2432,
+ serialized_end=2540,
)
_SCHEMA = _descriptor.Descriptor(
@@ -961,8 +969,8 @@ _SCHEMA = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=775,
- serialized_end=2797,
+ serialized_start=804,
+ serialized_end=2826,
)
@@ -1016,8 +1024,8 @@ _TYPEINFO_FIELDTYPE = _descriptor.Descriptor(
name='type_info', full_name='org.apache.flink.fn_execution.v1.TypeInfo.FieldType.type_info',
index=0, containing_type=None, fields=[]),
],
- serialized_start=2878,
- serialized_end=3203,
+ serialized_start=2907,
+ serialized_end=3232,
)
_TYPEINFO_FIELD = _descriptor.Descriptor(
@@ -1060,8 +1068,8 @@ _TYPEINFO_FIELD = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=3205,
- serialized_end=3315,
+ serialized_start=3234,
+ serialized_end=3344,
)
_TYPEINFO = _descriptor.Descriptor(
@@ -1091,8 +1099,8 @@ _TYPEINFO = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=2800,
- serialized_end=3549,
+ serialized_start=2829,
+ serialized_end=3578,
)
_USERDEFINEDFUNCTION_INPUT.fields_by_name['udf'].message_type = _USERDEFINEDFUNCTION
diff --git a/flink-python/pyflink/fn_execution/operation_utils.py b/flink-python/pyflink/fn_execution/operation_utils.py
index 52426ce..c361681 100644
--- a/flink-python/pyflink/fn_execution/operation_utils.py
+++ b/flink-python/pyflink/fn_execution/operation_utils.py
@@ -108,6 +108,12 @@ def extract_data_stream_stateless_funcs(udfs):
def wrap_func(value):
return reduce_func(value[0], value[1])
func = wrap_func
+ elif func_type == udf.CO_MAP:
+ co_map_func = cloudpickle.loads(udfs[0].payload)
+
+ def wrap_func(value):
+ return co_map_func.map1(value[1]) if value[0] else co_map_func.map2(value[2])
+ func = wrap_func
return func
diff --git a/flink-python/pyflink/proto/flink-fn-execution.proto b/flink-python/pyflink/proto/flink-fn-execution.proto
index 9a1e64b..cd903f3 100644
--- a/flink-python/pyflink/proto/flink-fn-execution.proto
+++ b/flink-python/pyflink/proto/flink-fn-execution.proto
@@ -58,6 +58,8 @@ message UserDefinedDataStreamFunction {
MAP = 0;
FLAT_MAP = 1;
REDUCE = 2;
+ CO_MAP = 3;
+ CO_FLAT_MAP = 4;
}
FunctionType functionType = 1;
bytes payload = 2;
diff --git a/flink-python/src/main/java/org/apache/flink/datastream/runtime/operators/python/DataStreamTwoInputPythonStatelessFunctionOperator.java b/flink-python/src/main/java/org/apache/flink/datastream/runtime/operators/python/DataStreamTwoInputPythonStatelessFunctionOperator.java
new file mode 100644
index 0000000..7383076
--- /dev/null
+++ b/flink-python/src/main/java/org/apache/flink/datastream/runtime/operators/python/DataStreamTwoInputPythonStatelessFunctionOperator.java
@@ -0,0 +1,196 @@
+/*
+ * 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.datastream.runtime.operators.python;
+
+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.typeutils.RowTypeInfo;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.memory.ByteArrayInputStreamWithPos;
+import org.apache.flink.core.memory.ByteArrayOutputStreamWithPos;
+import org.apache.flink.core.memory.DataInputViewStreamWrapper;
+import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
+import org.apache.flink.datastream.runtime.functions.python.DataStreamPythonFunctionInfo;
+import org.apache.flink.datastream.runtime.runners.python.beam.BeamDataStreamPythonStatelessFunctionRunner;
+import org.apache.flink.datastream.runtime.typeutils.python.PythonTypeUtils;
+import org.apache.flink.fnexecution.v1.FlinkFnApi;
+import org.apache.flink.python.PythonFunctionRunner;
+import org.apache.flink.streaming.api.operators.python.AbstractTwoInputPythonFunctionOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.table.functions.python.PythonEnv;
+import org.apache.flink.table.runtime.util.StreamRecordCollector;
+import org.apache.flink.types.Row;
+
+import com.google.protobuf.ByteString;
+
+import java.util.Map;
+
+/**
+ * {@link DataStreamTwoInputPythonStatelessFunctionOperator} is responsible for launching beam
+ * runner which will start a python harness to execute two-input user defined python function.
+ */
+public class DataStreamTwoInputPythonStatelessFunctionOperator<IN1, IN2, OUT>
+ extends AbstractTwoInputPythonFunctionOperator<IN1, IN2, OUT> {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final String DATA_STREAM_STATELESS_PYTHON_FUNCTION_URN =
+ "flink:transform:datastream_stateless_function:v1";
+ private static final String DATA_STREAM_MAP_FUNCTION_CODER_URN = "flink:coder:datastream:map_function:v1";
+ private static final String DATA_STREAM_FLAT_MAP_FUNCTION_CODER_URN = "flink:coder:datastream:flatmap_function:v1";
+
+
+ private final DataStreamPythonFunctionInfo pythonFunctionInfo;
+
+ private final TypeInformation<OUT> outputTypeInfo;
+
+ private final Map<String, String> jobOptions;
+ private transient TypeSerializer<OUT> outputTypeSerializer;
+
+ private transient ByteArrayInputStreamWithPos bais;
+
+ private transient DataInputViewStreamWrapper baisWrapper;
+
+ private transient ByteArrayOutputStreamWithPos baos;
+
+ private transient DataOutputViewStreamWrapper baosWrapper;
+
+ private transient StreamRecordCollector streamRecordCollector;
+
+ private transient TypeSerializer<Row> runnerInputTypeSerializer;
+
+ private final TypeInformation<Row> runnerInputTypeInfo;
+
+ private transient Row reuseRow;
+
+ public DataStreamTwoInputPythonStatelessFunctionOperator(
+ Configuration config,
+ TypeInformation<IN1> inputTypeInfo1,
+ TypeInformation<IN2> inputTypeInfo2,
+ TypeInformation<OUT> outputTypeInfo,
+ DataStreamPythonFunctionInfo pythonFunctionInfo) {
+ super(config);
+ this.pythonFunctionInfo = pythonFunctionInfo;
+ jobOptions = config.toMap();
+ this.outputTypeInfo = outputTypeInfo;
+ // The row contains three field. The first field indicate left input or right input
+ // The second field contains left input and the third field contains right input.
+ runnerInputTypeInfo = new RowTypeInfo(Types.BOOLEAN, inputTypeInfo1, inputTypeInfo2);
+ }
+
+ @Override
+ public void open() throws Exception {
+ super.open();
+ bais = new ByteArrayInputStreamWithPos();
+ baisWrapper = new DataInputViewStreamWrapper(bais);
+
+ baos = new ByteArrayOutputStreamWithPos();
+ baosWrapper = new DataOutputViewStreamWrapper(baos);
+ this.outputTypeSerializer = PythonTypeUtils.TypeInfoToSerializerConverter
+ .typeInfoSerializerConverter(outputTypeInfo);
+ runnerInputTypeSerializer = PythonTypeUtils.TypeInfoToSerializerConverter
+ .typeInfoSerializerConverter(runnerInputTypeInfo);
+
+ reuseRow = new Row(3);
+ this.streamRecordCollector = new StreamRecordCollector(output);
+ }
+
+ @Override
+ public PythonFunctionRunner createPythonFunctionRunner() throws Exception {
+
+ String coderUrn;
+ int functionType = this.pythonFunctionInfo.getFunctionType();
+ if (functionType == FlinkFnApi.UserDefinedDataStreamFunction.FunctionType.CO_MAP.getNumber()) {
+ coderUrn = DATA_STREAM_MAP_FUNCTION_CODER_URN;
+ } else if (functionType == FlinkFnApi.UserDefinedDataStreamFunction.FunctionType.CO_FLAT_MAP.getNumber()) {
+ coderUrn = DATA_STREAM_FLAT_MAP_FUNCTION_CODER_URN;
+ } else {
+ throw new RuntimeException("Function Type for ConnectedStream should be Map or FlatMap");
+ }
+
+ return new BeamDataStreamPythonStatelessFunctionRunner(
+ getRuntimeContext().getTaskName(),
+ createPythonEnvironmentManager(),
+ runnerInputTypeInfo,
+ outputTypeInfo,
+ DATA_STREAM_STATELESS_PYTHON_FUNCTION_URN,
+ getUserDefinedDataStreamFunctionsProto(),
+ coderUrn,
+ jobOptions,
+ getFlinkMetricContainer()
+ );
+ }
+
+ @Override
+ public PythonEnv getPythonEnv() {
+ return pythonFunctionInfo.getPythonFunction().getPythonEnv();
+ }
+
+ @Override
+ public void emitResult(Tuple2<byte[], Integer> resultTuple) throws Exception {
+ byte[] rawResult = resultTuple.f0;
+ int length = resultTuple.f1;
+ bais.setBuffer(rawResult, 0, length);
+ streamRecordCollector.collect(outputTypeSerializer.deserialize(baisWrapper));
+ }
+
+ @Override
+ public void processElement1(StreamRecord<IN1> element) throws Exception {
+ // construct combined row.
+ reuseRow.setField(0, true);
+ reuseRow.setField(1, element.getValue());
+ reuseRow.setField(2, null); // need to set null since it is a reuse row.
+ processElement();
+ }
+
+ @Override
+ public void processElement2(StreamRecord<IN2> element) throws Exception {
+ // construct combined row.
+ reuseRow.setField(0, false);
+ reuseRow.setField(1, null); // need to set null since it is a reuse row.
+ reuseRow.setField(2, element.getValue());
+ processElement();
+ }
+
+ private void processElement() throws Exception {
+ runnerInputTypeSerializer.serialize(reuseRow, baosWrapper);
+ pythonFunctionRunner.process(baos.toByteArray());
+ baos.reset();
+ checkInvokeFinishBundleByCount();
+ emitResults();
+ }
+
+ protected FlinkFnApi.UserDefinedDataStreamFunctions getUserDefinedDataStreamFunctionsProto() {
+ FlinkFnApi.UserDefinedDataStreamFunctions.Builder builder = FlinkFnApi.UserDefinedDataStreamFunctions.newBuilder();
+ builder.addUdfs(getUserDefinedDataStreamFunctionProto(pythonFunctionInfo));
+ return builder.build();
+ }
+
+ private FlinkFnApi.UserDefinedDataStreamFunction getUserDefinedDataStreamFunctionProto(
+ DataStreamPythonFunctionInfo dataStreamPythonFunctionInfo) {
+ FlinkFnApi.UserDefinedDataStreamFunction.Builder builder =
+ FlinkFnApi.UserDefinedDataStreamFunction.newBuilder();
+ builder.setFunctionType(FlinkFnApi.UserDefinedDataStreamFunction.FunctionType.forNumber(
+ dataStreamPythonFunctionInfo.getFunctionType()));
+ builder.setPayload(ByteString.copyFrom(
+ dataStreamPythonFunctionInfo.getPythonFunction().getSerializedPythonFunction()));
+ return builder.build();
+ }
+}
diff --git a/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractPythonFunctionOperator.java b/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractPythonFunctionOperator.java
index 7361320..1295b8f 100644
--- a/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractPythonFunctionOperator.java
+++ b/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractPythonFunctionOperator.java
@@ -19,340 +19,26 @@
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.configuration.Configuration;
-import org.apache.flink.configuration.MemorySize;
-import org.apache.flink.python.PythonConfig;
-import org.apache.flink.python.PythonFunctionRunner;
-import org.apache.flink.python.PythonOptions;
-import org.apache.flink.python.env.PythonDependencyInfo;
-import org.apache.flink.python.env.PythonEnvironmentManager;
-import org.apache.flink.python.env.beam.ProcessPythonEnvironmentManager;
-import org.apache.flink.python.metric.FlinkMetricContainer;
-import org.apache.flink.runtime.memory.MemoryManager;
-import org.apache.flink.runtime.memory.MemoryReservationException;
-import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
import org.apache.flink.streaming.api.operators.BoundedOneInput;
-import org.apache.flink.streaming.api.operators.ChainingStrategy;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
-import org.apache.flink.streaming.api.watermark.Watermark;
-import org.apache.flink.table.functions.python.PythonEnv;
-import org.apache.flink.util.Preconditions;
-
-import java.io.IOException;
-import java.util.concurrent.ScheduledFuture;
/**
- * Base class for all stream operators to execute Python functions.
+ * Base class for all one input stream operators to execute Python functions.
*/
@Internal
public abstract class AbstractPythonFunctionOperator<IN, OUT>
- extends AbstractStreamOperator<OUT>
+ extends AbstractPythonFunctionOperatorBase<OUT>
implements OneInputStreamOperator<IN, OUT>, BoundedOneInput {
private static final long serialVersionUID = 1L;
- /**
- * The {@link PythonFunctionRunner} which is responsible for Python user-defined function execution.
- */
- protected transient PythonFunctionRunner pythonFunctionRunner;
-
- /**
- * Max number of elements to include in a bundle.
- */
- protected transient int maxBundleSize;
-
- /**
- * Number of processed elements in the current bundle.
- */
- private transient int elementCount;
-
- /**
- * Max duration of a bundle.
- */
- private transient long maxBundleTimeMills;
-
- /**
- * Time that the last bundle was finished.
- */
- private transient long lastFinishBundleTime;
-
- /**
- * A timer that finishes the current bundle after a fixed amount of time.
- */
- private transient ScheduledFuture<?> checkFinishBundleTimer;
-
- /**
- * Callback to be executed after the current bundle was finished.
- */
- private transient Runnable bundleFinishedCallback;
-
- /**
- * The size of the reserved memory from the MemoryManager.
- */
- private transient long reservedMemory;
-
- /**
- * The python config.
- */
- private PythonConfig config;
-
public AbstractPythonFunctionOperator(Configuration config) {
- this.config = new PythonConfig(Preconditions.checkNotNull(config));
- this.chainingStrategy = ChainingStrategy.ALWAYS;
- }
-
- public PythonConfig getPythonConfig() {
- return config;
- }
-
- @Override
- public void open() throws Exception {
- try {
-
- if (config.isUsingManagedMemory()) {
- reserveMemoryForPythonWorker();
- }
-
- this.maxBundleSize = config.getMaxBundleSize();
- if (this.maxBundleSize <= 0) {
- this.maxBundleSize = PythonOptions.MAX_BUNDLE_SIZE.defaultValue();
- LOG.error("Invalid value for the maximum bundle size. Using default value of " +
- this.maxBundleSize + '.');
- } else {
- LOG.info("The maximum bundle size is configured to {}.", this.maxBundleSize);
- }
-
- this.maxBundleTimeMills = config.getMaxBundleTimeMills();
- if (this.maxBundleTimeMills <= 0L) {
- this.maxBundleTimeMills = PythonOptions.MAX_BUNDLE_TIME_MILLS.defaultValue();
- LOG.error("Invalid value for the maximum bundle time. Using default value of " +
- this.maxBundleTimeMills + '.');
- } else {
- LOG.info("The maximum bundle time is configured to {} milliseconds.", this.maxBundleTimeMills);
- }
-
- this.pythonFunctionRunner = createPythonFunctionRunner();
- this.pythonFunctionRunner.open(config);
-
- this.elementCount = 0;
- this.lastFinishBundleTime = getProcessingTimeService().getCurrentProcessingTime();
-
- // Schedule timer to check timeout of finish bundle.
- long bundleCheckPeriod = Math.max(this.maxBundleTimeMills, 1);
- this.checkFinishBundleTimer =
- getProcessingTimeService()
- .scheduleAtFixedRate(
- // ProcessingTimeService callbacks are executed under the checkpointing lock
- timestamp -> checkInvokeFinishBundleByTime(), bundleCheckPeriod, bundleCheckPeriod);
- } finally {
- super.open();
- }
- }
-
- @Override
- public void close() throws Exception {
- try {
- invokeFinishBundle();
- } finally {
- super.close();
- }
- }
-
- @Override
- public void dispose() throws Exception {
- try {
- if (checkFinishBundleTimer != null) {
- checkFinishBundleTimer.cancel(true);
- checkFinishBundleTimer = null;
- }
- if (pythonFunctionRunner != null) {
- pythonFunctionRunner.close();
- pythonFunctionRunner = null;
- }
- if (reservedMemory > 0) {
- getContainingTask().getEnvironment().getMemoryManager().releaseMemory(this, reservedMemory);
- reservedMemory = -1;
- }
- } finally {
- super.dispose();
- }
+ super(config);
}
@Override
public void endInput() throws Exception {
invokeFinishBundle();
}
-
- @Override
- public void prepareSnapshotPreBarrier(long checkpointId) throws Exception {
- try {
- invokeFinishBundle();
- } finally {
- super.prepareSnapshotPreBarrier(checkpointId);
- }
- }
-
- @Override
- public void processWatermark(Watermark mark) throws Exception {
- // Due to the asynchronous communication with the SDK harness,
- // a bundle might still be in progress and not all items have
- // yet been received from the SDK harness. If we just set this
- // watermark as the new output watermark, we could violate the
- // order of the records, i.e. pending items in the SDK harness
- // could become "late" although they were "on time".
- //
- // We can solve this problem using one of the following options:
- //
- // 1) Finish the current bundle and emit this watermark as the
- // new output watermark. Finishing the bundle ensures that
- // all the items have been processed by the SDK harness and
- // the execution results sent to the downstream operator.
- //
- // 2) Hold on the output watermark for as long as the current
- // bundle has not been finished. We have to remember to manually
- // finish the bundle in case we receive the final watermark.
- // To avoid latency, we should process this watermark again as
- // soon as the current bundle is finished.
- //
- // Approach 1) is the easiest and gives better latency, yet 2)
- // gives better throughput due to the bundle not getting cut on
- // every watermark. So we have implemented 2) below.
- if (mark.getTimestamp() == Long.MAX_VALUE) {
- invokeFinishBundle();
- super.processWatermark(mark);
- } else if (elementCount == 0) {
- // forward the watermark immediately if the bundle is already finished.
- super.processWatermark(mark);
- } else {
- // It is not safe to advance the output watermark yet, so add a hold on the current
- // output watermark.
- bundleFinishedCallback =
- () -> {
- try {
- // at this point the bundle is finished, allow the watermark to pass
- super.processWatermark(mark);
- } catch (Exception e) {
- throw new RuntimeException(
- "Failed to process watermark after finished bundle.", e);
- }
- };
- }
- }
-
- /**
- * Reset the {@link PythonConfig} if needed.
- * */
- public void setPythonConfig(PythonConfig pythonConfig) {
- this.config = pythonConfig;
- }
-
- /**
- * Returns the {@link PythonConfig}.
- * */
- public PythonConfig getConfig() {
- return config;
- }
-
- /**
- * Creates the {@link PythonFunctionRunner} which is responsible for Python user-defined function execution.
- */
- public abstract PythonFunctionRunner createPythonFunctionRunner() throws Exception;
-
- /**
- * Returns the {@link PythonEnv} used to create PythonEnvironmentManager..
- */
- public abstract PythonEnv getPythonEnv();
-
- /**
- * Sends the execution result to the downstream operator.
- */
- public abstract void emitResult(Tuple2<byte[], Integer> resultTuple) throws Exception;
-
- /**
- * Reserves the memory used by the Python worker from the MemoryManager. This makes sure that
- * the memory used by the Python worker is managed by Flink.
- */
- private void reserveMemoryForPythonWorker() throws MemoryReservationException {
- long requiredPythonWorkerMemory = MemorySize.parse(config.getPythonFrameworkMemorySize())
- .add(MemorySize.parse(config.getPythonDataBufferMemorySize()))
- .getBytes();
- MemoryManager memoryManager = getContainingTask().getEnvironment().getMemoryManager();
- long availableManagedMemory = memoryManager.computeMemorySize(
- getOperatorConfig().getManagedMemoryFraction());
- if (requiredPythonWorkerMemory <= availableManagedMemory) {
- memoryManager.reserveMemory(this, requiredPythonWorkerMemory);
- LOG.info("Reserved memory {} for Python worker.", requiredPythonWorkerMemory);
- this.reservedMemory = requiredPythonWorkerMemory;
- // TODO enforce the memory limit of the Python worker
- } else {
- LOG.warn("Required Python worker memory {} exceeds the available managed off-heap " +
- "memory {}. Skipping reserving off-heap memory from the MemoryManager. This does " +
- "not affect the functionality. However, it may affect the stability of a job as " +
- "the memory used by the Python worker is not managed by Flink.",
- requiredPythonWorkerMemory, availableManagedMemory);
- this.reservedMemory = -1;
- }
- }
-
- protected void emitResults() throws Exception {
- Tuple2<byte[], Integer> resultTuple;
- while ((resultTuple = pythonFunctionRunner.pollResult()) != null) {
- emitResult(resultTuple);
- }
- }
-
- /**
- * Checks whether to invoke finishBundle by elements count. Called in processElement.
- */
- protected void checkInvokeFinishBundleByCount() throws Exception {
- elementCount++;
- if (elementCount >= maxBundleSize) {
- invokeFinishBundle();
- }
- }
-
- /**
- * Checks whether to invoke finishBundle by timeout.
- */
- private void checkInvokeFinishBundleByTime() throws Exception {
- long now = getProcessingTimeService().getCurrentProcessingTime();
- if (now - lastFinishBundleTime >= maxBundleTimeMills) {
- invokeFinishBundle();
- }
- }
-
- protected void invokeFinishBundle() throws Exception {
- if (elementCount > 0) {
- pythonFunctionRunner.flush();
- elementCount = 0;
- emitResults();
- lastFinishBundleTime = getProcessingTimeService().getCurrentProcessingTime();
- // callback only after current bundle was fully finalized
- if (bundleFinishedCallback != null) {
- bundleFinishedCallback.run();
- bundleFinishedCallback = null;
- }
- }
- }
-
- protected PythonEnvironmentManager createPythonEnvironmentManager() throws IOException {
- PythonDependencyInfo dependencyInfo = PythonDependencyInfo.create(
- config, getRuntimeContext().getDistributedCache());
- PythonEnv pythonEnv = getPythonEnv();
- if (pythonEnv.getExecType() == PythonEnv.ExecType.PROCESS) {
- return new ProcessPythonEnvironmentManager(
- dependencyInfo,
- getContainingTask().getEnvironment().getTaskManagerInfo().getTmpDirectories(),
- System.getenv());
- } else {
- throw new UnsupportedOperationException(String.format(
- "Execution type '%s' is not supported.", pythonEnv.getExecType()));
- }
- }
-
- protected FlinkMetricContainer getFlinkMetricContainer() {
- return this.config.isMetricEnabled() ?
- new FlinkMetricContainer(getRuntimeContext().getMetricGroup()) : null;
- }
}
diff --git a/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractPythonFunctionOperator.java b/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractPythonFunctionOperatorBase.java
similarity index 95%
copy from flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractPythonFunctionOperator.java
copy to flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractPythonFunctionOperatorBase.java
index 7361320..280bf0a 100644
--- a/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractPythonFunctionOperator.java
+++ b/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractPythonFunctionOperatorBase.java
@@ -32,9 +32,7 @@ import org.apache.flink.python.metric.FlinkMetricContainer;
import org.apache.flink.runtime.memory.MemoryManager;
import org.apache.flink.runtime.memory.MemoryReservationException;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
-import org.apache.flink.streaming.api.operators.BoundedOneInput;
import org.apache.flink.streaming.api.operators.ChainingStrategy;
-import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.table.functions.python.PythonEnv;
import org.apache.flink.util.Preconditions;
@@ -46,9 +44,8 @@ import java.util.concurrent.ScheduledFuture;
* Base class for all stream operators to execute Python functions.
*/
@Internal
-public abstract class AbstractPythonFunctionOperator<IN, OUT>
- extends AbstractStreamOperator<OUT>
- implements OneInputStreamOperator<IN, OUT>, BoundedOneInput {
+public abstract class AbstractPythonFunctionOperatorBase<OUT>
+ extends AbstractStreamOperator<OUT> {
private static final long serialVersionUID = 1L;
@@ -65,7 +62,7 @@ public abstract class AbstractPythonFunctionOperator<IN, OUT>
/**
* Number of processed elements in the current bundle.
*/
- private transient int elementCount;
+ protected transient int elementCount;
/**
* Max duration of a bundle.
@@ -85,7 +82,7 @@ public abstract class AbstractPythonFunctionOperator<IN, OUT>
/**
* Callback to be executed after the current bundle was finished.
*/
- private transient Runnable bundleFinishedCallback;
+ protected transient Runnable bundleFinishedCallback;
/**
* The size of the reserved memory from the MemoryManager.
@@ -97,7 +94,7 @@ public abstract class AbstractPythonFunctionOperator<IN, OUT>
*/
private PythonConfig config;
- public AbstractPythonFunctionOperator(Configuration config) {
+ public AbstractPythonFunctionOperatorBase(Configuration config) {
this.config = new PythonConfig(Preconditions.checkNotNull(config));
this.chainingStrategy = ChainingStrategy.ALWAYS;
}
@@ -180,11 +177,6 @@ public abstract class AbstractPythonFunctionOperator<IN, OUT>
}
@Override
- public void endInput() throws Exception {
- invokeFinishBundle();
- }
-
- @Override
public void prepareSnapshotPreBarrier(long checkpointId) throws Exception {
try {
invokeFinishBundle();
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
new file mode 100644
index 0000000..ed221c6
--- /dev/null
+++ b/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractTwoInputPythonFunctionOperator.java
@@ -0,0 +1,44 @@
+/*
+ * 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.operators.python;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.operators.BoundedMultiInput;
+import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
+
+/**
+ * Base class for all two input stream operators to execute Python functions.
+ */
+@Internal
+public abstract class AbstractTwoInputPythonFunctionOperator<IN1, IN2, OUT>
+ extends AbstractPythonFunctionOperatorBase<OUT>
+ implements TwoInputStreamOperator<IN1, IN2, OUT>, BoundedMultiInput {
+
+ private static final long serialVersionUID = 1L;
+
+ public AbstractTwoInputPythonFunctionOperator(Configuration config) {
+ super(config);
+ }
+
+ @Override
+ public void endInput(int inputId) throws Exception {
+ invokeFinishBundle();
+ }
+}