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 2020/10/15 01:37:52 UTC
[flink] branch master updated: [FLINK-19480][python] Support
RichFunction in Python DataStream API
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
The following commit(s) were added to refs/heads/master by this push:
new c8938f9 [FLINK-19480][python] Support RichFunction in Python DataStream API
c8938f9 is described below
commit c8938f9e238f8eb05430386d5bc3104b23d6c37c
Author: Dian Fu <di...@apache.org>
AuthorDate: Tue Oct 13 08:50:10 2020 +0800
[FLINK-19480][python] Support RichFunction in Python DataStream API
This closes #13611.
---
flink-python/pyflink/datastream/data_stream.py | 20 +-
flink-python/pyflink/datastream/functions.py | 80 ++++++-
.../fn_execution/beam/beam_operations_fast.pyx | 52 +++--
.../fn_execution/beam/beam_operations_slow.py | 55 +++--
.../pyflink/fn_execution/flink_fn_execution_pb2.py | 260 ++++++++++++++-------
.../pyflink/fn_execution/operation_utils.py | 37 ++-
.../pyflink/proto/flink-fn-execution.proto | 26 ++-
.../python/PythonPartitionCustomOperator.java | 15 +-
.../api/operators/python/PythonReduceOperator.java | 8 +-
.../StatelessOneInputPythonFunctionOperator.java | 26 +--
.../StatelessTwoInputPythonFunctionOperator.java | 26 +--
...eamDataStreamStatelessPythonFunctionRunner.java | 10 +-
.../api}/utils/PythonOperatorUtils.java | 46 +++-
.../api/{typeutils => utils}/PythonTypeUtils.java | 2 +-
.../AbstractPythonScalarFunctionFlatMap.java | 2 +-
.../python/PythonTableFunctionFlatMap.java | 2 +-
.../PythonStreamGroupAggregateOperator.java | 2 +-
...stractArrowPythonAggregateFunctionOperator.java | 2 +-
...wPythonOverWindowAggregateFunctionOperator.java | 2 +-
.../AbstractPythonScalarFunctionOperator.java | 2 +-
.../table/AbstractPythonTableFunctionOperator.java | 2 +-
.../{typeutils => utils}/PythonTypeUtilsTest.java | 2 +-
22 files changed, 453 insertions(+), 226 deletions(-)
diff --git a/flink-python/pyflink/datastream/data_stream.py b/flink-python/pyflink/datastream/data_stream.py
index a1c5cb5..74aac0a 100644
--- a/flink-python/pyflink/datastream/data_stream.py
+++ b/flink-python/pyflink/datastream/data_stream.py
@@ -15,7 +15,6 @@
# See the License for the specific language governing permissions and
# limitations under the License.
################################################################################
-import os
from typing import Callable, Union
from pyflink.common import typeinfo, ExecutionConfig, Row
@@ -25,7 +24,7 @@ from pyflink.datastream.functions import _get_python_env, FlatMapFunctionWrapper
MapFunction, MapFunctionWrapper, Function, FunctionWrapper, SinkFunction, FilterFunction, \
FilterFunctionWrapper, KeySelectorFunctionWrapper, KeySelector, ReduceFunction, \
ReduceFunctionWrapper, CoMapFunction, CoFlatMapFunction, Partitioner, \
- PartitionerFunctionWrapper
+ PartitionerFunctionWrapper, RuntimeContext
from pyflink.java_gateway import get_gateway
@@ -467,25 +466,26 @@ class DataStream(object):
raise TypeError("Parameter partitioner should be a type of Partitioner.")
gateway = get_gateway()
- data_stream_num_partitions_env_key = \
- gateway.jvm.PythonPartitionCustomOperator.DATA_STREAM_NUM_PARTITIONS
class PartitionCustomMapFunction(MapFunction):
"""
A wrapper class for partition_custom map function. It indicates that it is a partition
- custom operation that we need to apply DataStreamPythonPartitionCustomFunctionOperator
+ custom operation that we need to apply PythonPartitionCustomOperator
to run the map function.
"""
def __init__(self):
self.num_partitions = None
- def map(self, value):
- return self.partition_custom_map(value)
+ def open(self, runtime_context: RuntimeContext):
+ self.num_partitions = int(runtime_context.get_job_parameter(
+ "DATA_STREAM_NUM_PARTITIONS", "-1"))
+ if self.num_partitions <= 0:
+ raise ValueError(
+ "The partition number should be a positive value, got %s"
+ % self.num_partitions)
- def partition_custom_map(self, value):
- if self.num_partitions is None:
- self.num_partitions = int(os.environ[data_stream_num_partitions_env_key])
+ def map(self, value):
partition = partitioner.partition(key_selector.get_key(value), self.num_partitions)
return Row(partition, value)
diff --git a/flink-python/pyflink/datastream/functions.py b/flink-python/pyflink/datastream/functions.py
index 5402062..8aefafc 100644
--- a/flink-python/pyflink/datastream/functions.py
+++ b/flink-python/pyflink/datastream/functions.py
@@ -17,18 +17,92 @@
################################################################################
import abc
-from typing import Union, Any
+from typing import Union, Any, Dict
from py4j.java_gateway import JavaObject
from pyflink.java_gateway import get_gateway
+class RuntimeContext(object):
+ """
+ A RuntimeContext contains information about the context in which functions are executed.
+ Each parallel instance of the function will have a context through which it can access
+ static contextual information (such as the current parallelism).
+ """
+
+ def __init__(self,
+ task_name: str,
+ task_name_with_subtasks: str,
+ number_of_parallel_subtasks: int,
+ max_number_of_parallel_subtasks: int,
+ index_of_this_subtask: int,
+ attempt_number: int,
+ job_parameters: Dict[str, str]):
+ self._task_name = task_name
+ self._task_name_with_subtasks = task_name_with_subtasks
+ self._number_of_parallel_subtasks = number_of_parallel_subtasks
+ self._max_number_of_parallel_subtasks = max_number_of_parallel_subtasks
+ self._index_of_this_subtask = index_of_this_subtask
+ self._attempt_number = attempt_number
+ self._job_parameters = job_parameters
+
+ def get_task_name(self) -> str:
+ """
+ Returns the name of the task in which the UDF runs, as assigned during plan construction.
+ """
+ return self._task_name
+
+ def get_number_of_parallel_subtasks(self) -> int:
+ """
+ Gets the parallelism with which the parallel task runs.
+ """
+ return self._number_of_parallel_subtasks
+
+ def get_max_number_of_parallel_subtasks(self) -> int:
+ """
+ Gets the number of max-parallelism with which the parallel task runs.
+ """
+ return self._max_number_of_parallel_subtasks
+
+ def get_index_of_this_subtask(self) -> int:
+ """
+ Gets the number of this parallel subtask. The numbering starts from 0 and goes up to
+ parallelism-1 (parallelism as returned by
+ :func:`~RuntimeContext.get_number_of_parallel_subtasks`).
+ """
+ return self._index_of_this_subtask
+
+ def get_attempt_number(self) -> int:
+ """
+ Gets the attempt number of this parallel subtask. First attempt is numbered 0.
+ """
+ return self._attempt_number
+
+ def get_task_name_with_subtasks(self) -> str:
+ """
+ Returns the name of the task, appended with the subtask indicator, such as "MyTask (3/6)",
+ where 3 would be (:func:`~RuntimeContext.get_index_of_this_subtask` + 1), and 6 would be
+ :func:`~RuntimeContext.get_number_of_parallel_subtasks`.
+ """
+ return self._task_name_with_subtasks
+
+ def get_job_parameter(self, key: str, default_value: str):
+ """
+ Gets the global job parameter value associated with the given key as a string.
+ """
+ return self._job_parameters[key] if key in self._job_parameters else default_value
+
+
class Function(abc.ABC):
"""
The base class for all user-defined functions.
"""
- pass
+ def open(self, runtime_context: RuntimeContext):
+ pass
+
+ def close(self):
+ pass
class MapFunction(Function):
@@ -266,7 +340,7 @@ class Partitioner(Function):
pass
-class FunctionWrapper(object):
+class FunctionWrapper(Function):
"""
A basic wrapper class for user defined function.
"""
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 94238bc..56f01ff 100644
--- a/flink-python/pyflink/fn_execution/beam/beam_operations_fast.pyx
+++ b/flink-python/pyflink/fn_execution/beam/beam_operations_fast.pyx
@@ -28,6 +28,7 @@ from apache_beam.runners.worker import bundle_processor
from apache_beam.runners.worker import operation_specs
from apache_beam.utils.windowed_value cimport WindowedValue
+from pyflink.datastream.functions import RuntimeContext
from pyflink.fn_execution.coder_impl_fast cimport BaseCoderImpl
from pyflink.fn_execution.beam.beam_stream cimport BeamInputStream, BeamOutputStream
from pyflink.fn_execution.beam.beam_coder_impl_fast cimport InputStreamWrapper
@@ -66,7 +67,7 @@ cdef class BeamStatelessFunctionOperation(Operation):
self._is_python_coder = False
self._output_coder = self._value_coder_impl._value_coder
- self.func, self.user_defined_funcs = self.generate_func(self.spec.serialized_fn.udfs)
+ self.func, self.user_defined_funcs = self.generate_func(self.spec.serialized_fn)
self._metric_enabled = self.spec.serialized_fn.metric_enabled
self.base_metric_group = None
if self._metric_enabled:
@@ -77,7 +78,7 @@ cdef class BeamStatelessFunctionOperation(Operation):
for user_defined_func in self.user_defined_funcs:
user_defined_func.open(FunctionContext(self.base_metric_group))
- def generate_func(self, udfs) -> tuple:
+ def generate_func(self, serialized_fn) -> tuple:
pass
cpdef start(self):
@@ -145,10 +146,11 @@ cdef class BeamScalarFunctionOperation(BeamStatelessFunctionOperation):
super(BeamScalarFunctionOperation, self).__init__(
name, spec, counter_factory, sampler, consumers)
- def generate_func(self, udfs):
+ def generate_func(self, serialized_fn):
"""
Generates a lambda function based on udfs.
- :param udfs: a list of the proto representation of the Python :class:`ScalarFunction`
+ :param serialized_fn: serialized function which contains a list of the proto
+ representation of the Python :class:`ScalarFunction`
:return: the generated lambda function
"""
scalar_functions, variable_dict, user_defined_funcs = reduce(
@@ -156,7 +158,7 @@ cdef class BeamScalarFunctionOperation(BeamStatelessFunctionOperation):
','.join([x[0], y[0]]),
dict(chain(x[1].items(), y[1].items())),
x[2] + y[2]),
- [operation_utils.extract_user_defined_function(udf) for udf in udfs])
+ [operation_utils.extract_user_defined_function(udf) for udf in serialized_fn.udfs])
mapper = eval('lambda value: [%s]' % scalar_functions, variable_dict)
if self._is_python_coder:
generate_func = lambda it: map(mapper, it)
@@ -169,14 +171,14 @@ cdef class BeamTableFunctionOperation(BeamStatelessFunctionOperation):
super(BeamTableFunctionOperation, self).__init__(
name, spec, counter_factory, sampler, consumers)
- def generate_func(self, udtfs):
+ def generate_func(self, serialized_fn):
"""
Generates a lambda function based on udtfs.
:param udtfs: a list of the proto representation of the Python :class:`TableFunction`
:return: the generated lambda function
"""
table_function, variable_dict, user_defined_funcs = \
- operation_utils.extract_user_defined_function(udtfs[0])
+ operation_utils.extract_user_defined_function(serialized_fn.udfs[0])
generate_func = eval('lambda value: %s' % table_function, variable_dict)
return generate_func, user_defined_funcs
@@ -185,9 +187,21 @@ cdef class DataStreamStatelessFunctionOperation(BeamStatelessFunctionOperation):
super(DataStreamStatelessFunctionOperation, self).__init__(name, spec, counter_factory,
sampler, consumers)
- def generate_func(self, udfs):
- func = operation_utils.extract_data_stream_stateless_funcs(udfs)
- return func, []
+ def open_func(self):
+ for user_defined_func in self.user_defined_funcs:
+ runtime_context = RuntimeContext(
+ self.spec.serialized_fn.runtime_context.task_name,
+ self.spec.serialized_fn.runtime_context.task_name_with_subtasks,
+ self.spec.serialized_fn.runtime_context.number_of_parallel_subtasks,
+ self.spec.serialized_fn.runtime_context.max_number_of_parallel_subtasks,
+ self.spec.serialized_fn.runtime_context.index_of_this_subtask,
+ self.spec.serialized_fn.runtime_context.attempt_number,
+ {p.key: p.value for p in self.spec.serialized_fn.runtime_context.job_parameters})
+ user_defined_func.open(runtime_context)
+
+ def generate_func(self, serialized_fn):
+ func, user_defined_func = operation_utils.extract_data_stream_stateless_funcs(serialized_fn)
+ return func, [user_defined_func]
cdef class PandasAggregateFunctionOperation(BeamStatelessFunctionOperation):
@@ -195,13 +209,14 @@ cdef class PandasAggregateFunctionOperation(BeamStatelessFunctionOperation):
super(PandasAggregateFunctionOperation, self).__init__(name, spec, counter_factory,
sampler, consumers)
- def generate_func(self, udfs):
+ def generate_func(self, serialized_fn):
pandas_functions, variable_dict, user_defined_funcs = reduce(
lambda x, y: (
','.join([x[0], y[0]]),
dict(chain(x[1].items(), y[1].items())),
x[2] + y[2]),
- [operation_utils.extract_user_defined_function(udf, True) for udf in udfs])
+ [operation_utils.extract_user_defined_function(udf, True)
+ for udf in serialized_fn.udfs])
variable_dict['wrap_pandas_result'] = operation_utils.wrap_pandas_result
mapper = eval('lambda value: wrap_pandas_result([%s])' % pandas_functions, variable_dict)
generate_func = lambda it: map(mapper, it)
@@ -228,11 +243,11 @@ cdef class PandasBatchOverWindowAggregateFunctionOperation(BeamStatelessFunction
else:
self.is_bounded_range_window.append(False)
- def generate_func(self, udfs):
+ def generate_func(self, serialized_fn):
user_defined_funcs = []
self.window_indexes = []
self.mapper = []
- for udf in udfs:
+ for udf in serialized_fn.udfs:
pandas_agg_function, variable_dict, user_defined_func, window_index = \
operation_utils.extract_over_window_user_defined_function(udf)
user_defined_funcs.extend(user_defined_func)
@@ -343,12 +358,13 @@ class BeamStreamGroupAggregateOperation(BeamStatefulFunctionOperation):
def open_func(self):
self.group_agg_function.open(FunctionContext(self.base_metric_group))
- def generate_func(self, udfs):
+ def generate_func(self, serialized_fn):
user_defined_aggs = []
input_extractors = []
- for i in range(len(udfs)):
+ for i in range(len(serialized_fn.udfs)):
if i != self.index_of_count_star:
- user_defined_agg, input_extractor = extract_user_defined_aggregate_function(udfs[i])
+ user_defined_agg, input_extractor = extract_user_defined_aggregate_function(
+ serialized_fn.udfs[i])
else:
user_defined_agg = Count1AggFunction()
input_extractor = lambda value: []
@@ -403,7 +419,7 @@ def create_table_function(factory, transform_id, transform_proto, parameter, con
factory, transform_proto, consumers, parameter, BeamTableFunctionOperation)
@bundle_processor.BeamTransformFactory.register_urn(
- operation_utils.DATA_STREAM_STATELESS_FUNCTION_URN, flink_fn_execution_pb2.UserDefinedDataStreamFunctions)
+ operation_utils.DATA_STREAM_STATELESS_FUNCTION_URN, flink_fn_execution_pb2.UserDefinedDataStreamFunction)
def create_data_stream_function(factory, transform_id, transform_proto, parameter, consumers):
return _create_user_defined_function_operation(
factory, transform_proto, consumers, parameter, DataStreamStatelessFunctionOperation)
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 d95316a..e4b83b5 100644
--- a/flink-python/pyflink/fn_execution/beam/beam_operations_slow.py
+++ b/flink-python/pyflink/fn_execution/beam/beam_operations_slow.py
@@ -25,6 +25,7 @@ from apache_beam.runners.worker.operations import Operation
from apache_beam.utils.windowed_value import WindowedValue
from typing import Tuple
+from pyflink.datastream.functions import RuntimeContext
from pyflink.fn_execution import flink_fn_execution_pb2, operation_utils
from pyflink.fn_execution.beam.beam_coders import DataViewFilterCoder
from pyflink.fn_execution.coders import from_proto
@@ -48,7 +49,7 @@ class StatelessFunctionOperation(Operation):
self.consumer = consumers['output'][0]
self._value_coder_impl = self.consumer.windowed_coder.wrapped_value_coder.get_impl()
- self.func, self.user_defined_funcs = self.generate_func(self.spec.serialized_fn.udfs)
+ self.func, self.user_defined_funcs = self.generate_func(self.spec.serialized_fn)
self._metric_enabled = self.spec.serialized_fn.metric_enabled
self.base_metric_group = None
if self._metric_enabled:
@@ -104,7 +105,7 @@ class StatelessFunctionOperation(Operation):
"""
return super().user_monitoring_infos(transform_id)
- def generate_func(self, udfs) -> tuple:
+ def generate_func(self, serialized_fn) -> tuple:
pass
@staticmethod
@@ -123,10 +124,11 @@ class ScalarFunctionOperation(StatelessFunctionOperation):
super(ScalarFunctionOperation, self).__init__(
name, spec, counter_factory, sampler, consumers)
- def generate_func(self, udfs):
+ def generate_func(self, serialized_fn):
"""
Generates a lambda function based on udfs.
- :param udfs: a list of the proto representation of the Python :class:`ScalarFunction`
+ :param serialized_fn: serialized function which contains a list of the proto
+ representation of the Python :class:`ScalarFunction`
:return: the generated lambda function
"""
scalar_functions, variable_dict, user_defined_funcs = reduce(
@@ -134,7 +136,7 @@ class ScalarFunctionOperation(StatelessFunctionOperation):
','.join([x[0], y[0]]),
dict(chain(x[1].items(), y[1].items())),
x[2] + y[2]),
- [operation_utils.extract_user_defined_function(udf) for udf in udfs])
+ [operation_utils.extract_user_defined_function(udf) for udf in serialized_fn.udfs])
mapper = eval('lambda value: [%s]' % scalar_functions, variable_dict)
return lambda it: map(mapper, it), user_defined_funcs
@@ -144,14 +146,15 @@ class TableFunctionOperation(StatelessFunctionOperation):
super(TableFunctionOperation, self).__init__(
name, spec, counter_factory, sampler, consumers)
- def generate_func(self, udtfs):
+ def generate_func(self, serialized_fn):
"""
Generates a lambda function based on udtfs.
- :param udtfs: a list of the proto representation of the Python :class:`TableFunction`
+ :param serialized_fn: serialized function which contains the proto representation of
+ the Python :class:`TableFunction`
:return: the generated lambda function
"""
table_function, variable_dict, user_defined_funcs = \
- operation_utils.extract_user_defined_function(udtfs[0])
+ operation_utils.extract_user_defined_function(serialized_fn.udfs[0])
mapper = eval('lambda value: %s' % table_function, variable_dict)
return lambda it: map(mapper, it), user_defined_funcs
@@ -162,9 +165,21 @@ class DataStreamStatelessFunctionOperation(StatelessFunctionOperation):
super(DataStreamStatelessFunctionOperation, self).__init__(name, spec, counter_factory,
sampler, consumers)
- def generate_func(self, udfs):
- func = operation_utils.extract_data_stream_stateless_funcs(udfs=udfs)
- return lambda it: map(func, it), []
+ def open_func(self):
+ for user_defined_func in self.user_defined_funcs:
+ runtime_context = RuntimeContext(
+ self.spec.serialized_fn.runtime_context.task_name,
+ self.spec.serialized_fn.runtime_context.task_name_with_subtasks,
+ self.spec.serialized_fn.runtime_context.number_of_parallel_subtasks,
+ self.spec.serialized_fn.runtime_context.max_number_of_parallel_subtasks,
+ self.spec.serialized_fn.runtime_context.index_of_this_subtask,
+ self.spec.serialized_fn.runtime_context.attempt_number,
+ {p.key: p.value for p in self.spec.serialized_fn.runtime_context.job_parameters})
+ user_defined_func.open(runtime_context)
+
+ def generate_func(self, serialized_fn):
+ func, user_defined_func = operation_utils.extract_data_stream_stateless_funcs(serialized_fn)
+ return lambda it: map(func, it), [user_defined_func]
class PandasAggregateFunctionOperation(StatelessFunctionOperation):
@@ -172,13 +187,14 @@ class PandasAggregateFunctionOperation(StatelessFunctionOperation):
super(PandasAggregateFunctionOperation, self).__init__(
name, spec, counter_factory, sampler, consumers)
- def generate_func(self, udfs):
+ def generate_func(self, serialized_fn):
pandas_functions, variable_dict, user_defined_funcs = reduce(
lambda x, y: (
','.join([x[0], y[0]]),
dict(chain(x[1].items(), y[1].items())),
x[2] + y[2]),
- [operation_utils.extract_user_defined_function(udf, True) for udf in udfs])
+ [operation_utils.extract_user_defined_function(udf, True)
+ for udf in serialized_fn.udfs])
variable_dict['wrap_pandas_result'] = operation_utils.wrap_pandas_result
mapper = eval('lambda value: wrap_pandas_result([%s])' % pandas_functions, variable_dict)
return lambda it: map(mapper, it), user_defined_funcs
@@ -207,11 +223,11 @@ class PandasBatchOverWindowAggregateFunctionOperation(StatelessFunctionOperation
else:
self.is_bounded_range_window.append(False)
- def generate_func(self, udfs):
+ def generate_func(self, serialized_fn):
user_defined_funcs = []
self.window_indexes = []
self.mapper = []
- for udf in udfs:
+ for udf in serialized_fn.udfs:
pandas_agg_function, variable_dict, user_defined_func, window_index = \
operation_utils.extract_over_window_user_defined_function(udf)
user_defined_funcs.extend(user_defined_func)
@@ -327,12 +343,13 @@ class StreamGroupAggregateOperation(StatefulFunctionOperation):
def open_func(self):
self.group_agg_function.open(FunctionContext(self.base_metric_group))
- def generate_func(self, udfs):
+ def generate_func(self, serialized_fn):
user_defined_aggs = []
input_extractors = []
- for i in range(len(udfs)):
+ for i in range(len(serialized_fn.udfs)):
if i != self.index_of_count_star:
- user_defined_agg, input_extractor = extract_user_defined_aggregate_function(udfs[i])
+ user_defined_agg, input_extractor = extract_user_defined_aggregate_function(
+ serialized_fn.udfs[i])
else:
user_defined_agg = Count1AggFunction()
@@ -393,7 +410,7 @@ def create_table_function(factory, transform_id, transform_proto, parameter, con
@bundle_processor.BeamTransformFactory.register_urn(
operation_utils.DATA_STREAM_STATELESS_FUNCTION_URN,
- flink_fn_execution_pb2.UserDefinedDataStreamFunctions)
+ flink_fn_execution_pb2.UserDefinedDataStreamFunction)
def create_data_stream_function(factory, transform_id, transform_proto, parameter, consumers):
return _create_user_defined_function_operation(
factory, transform_proto, consumers, parameter, DataStreamStatelessFunctionOperation)
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 3f31c639..2b40b92 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\"u\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\n\x0cw [...]
+ 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\"u\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\n\x0cw [...]
)
@@ -116,8 +116,8 @@ _USERDEFINEDDATASTREAMFUNCTION_FUNCTIONTYPE = _descriptor.EnumDescriptor(
],
containing_type=None,
options=None,
- serialized_start=1002,
- serialized_end=1080,
+ serialized_start=1482,
+ serialized_end=1560,
)
_sym_db.RegisterEnumDescriptor(_USERDEFINEDDATASTREAMFUNCTION_FUNCTIONTYPE)
@@ -214,8 +214,8 @@ _SCHEMA_TYPENAME = _descriptor.EnumDescriptor(
],
containing_type=None,
options=None,
- serialized_start=4019,
- serialized_end=4308,
+ serialized_start=4361,
+ serialized_end=4650,
)
_sym_db.RegisterEnumDescriptor(_SCHEMA_TYPENAME)
@@ -304,8 +304,8 @@ _TYPEINFO_TYPENAME = _descriptor.EnumDescriptor(
],
containing_type=None,
options=None,
- serialized_start=4829,
- serialized_end=5087,
+ serialized_start=5171,
+ serialized_end=5429,
)
_sym_db.RegisterEnumDescriptor(_TYPEINFO_TYPENAME)
@@ -494,24 +494,24 @@ _OVERWINDOW = _descriptor.Descriptor(
)
-_USERDEFINEDDATASTREAMFUNCTION = _descriptor.Descriptor(
- name='UserDefinedDataStreamFunction',
- full_name='org.apache.flink.fn_execution.v1.UserDefinedDataStreamFunction',
+_USERDEFINEDDATASTREAMFUNCTION_JOBPARAMETER = _descriptor.Descriptor(
+ name='JobParameter',
+ full_name='org.apache.flink.fn_execution.v1.UserDefinedDataStreamFunction.JobParameter',
filename=None,
file=DESCRIPTOR,
containing_type=None,
fields=[
_descriptor.FieldDescriptor(
- name='functionType', full_name='org.apache.flink.fn_execution.v1.UserDefinedDataStreamFunction.functionType', index=0,
- number=1, type=14, cpp_type=8, label=1,
- has_default_value=False, default_value=0,
+ name='key', full_name='org.apache.flink.fn_execution.v1.UserDefinedDataStreamFunction.JobParameter.key', index=0,
+ number=1, type=9, cpp_type=9, label=1,
+ has_default_value=False, default_value=_b("").decode('utf-8'),
message_type=None, enum_type=None, containing_type=None,
is_extension=False, extension_scope=None,
options=None, file=DESCRIPTOR),
_descriptor.FieldDescriptor(
- name='payload', full_name='org.apache.flink.fn_execution.v1.UserDefinedDataStreamFunction.payload', index=1,
- number=2, type=12, cpp_type=9, label=1,
- has_default_value=False, default_value=_b(""),
+ name='value', full_name='org.apache.flink.fn_execution.v1.UserDefinedDataStreamFunction.JobParameter.value', index=1,
+ number=2, type=9, cpp_type=9, label=1,
+ has_default_value=False, default_value=_b("").decode('utf-8'),
message_type=None, enum_type=None, containing_type=None,
is_extension=False, extension_scope=None,
options=None, file=DESCRIPTOR),
@@ -520,7 +520,6 @@ _USERDEFINEDDATASTREAMFUNCTION = _descriptor.Descriptor(
],
nested_types=[],
enum_types=[
- _USERDEFINEDDATASTREAMFUNCTION_FUNCTIONTYPE,
],
options=None,
is_extendable=False,
@@ -528,28 +527,113 @@ _USERDEFINEDDATASTREAMFUNCTION = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=852,
- serialized_end=1080,
+ serialized_start=1132,
+ serialized_end=1174,
)
-
-_USERDEFINEDDATASTREAMFUNCTIONS = _descriptor.Descriptor(
- name='UserDefinedDataStreamFunctions',
- full_name='org.apache.flink.fn_execution.v1.UserDefinedDataStreamFunctions',
+_USERDEFINEDDATASTREAMFUNCTION_RUNTIMECONTEXT = _descriptor.Descriptor(
+ name='RuntimeContext',
+ full_name='org.apache.flink.fn_execution.v1.UserDefinedDataStreamFunction.RuntimeContext',
filename=None,
file=DESCRIPTOR,
containing_type=None,
fields=[
_descriptor.FieldDescriptor(
- name='udfs', full_name='org.apache.flink.fn_execution.v1.UserDefinedDataStreamFunctions.udfs', index=0,
- number=1, type=11, cpp_type=10, label=3,
+ name='task_name', full_name='org.apache.flink.fn_execution.v1.UserDefinedDataStreamFunction.RuntimeContext.task_name', index=0,
+ number=1, type=9, cpp_type=9, label=1,
+ has_default_value=False, default_value=_b("").decode('utf-8'),
+ message_type=None, enum_type=None, containing_type=None,
+ is_extension=False, extension_scope=None,
+ options=None, file=DESCRIPTOR),
+ _descriptor.FieldDescriptor(
+ name='task_name_with_subtasks', full_name='org.apache.flink.fn_execution.v1.UserDefinedDataStreamFunction.RuntimeContext.task_name_with_subtasks', index=1,
+ number=2, type=9, cpp_type=9, label=1,
+ has_default_value=False, default_value=_b("").decode('utf-8'),
+ message_type=None, enum_type=None, containing_type=None,
+ is_extension=False, extension_scope=None,
+ options=None, file=DESCRIPTOR),
+ _descriptor.FieldDescriptor(
+ name='number_of_parallel_subtasks', full_name='org.apache.flink.fn_execution.v1.UserDefinedDataStreamFunction.RuntimeContext.number_of_parallel_subtasks', index=2,
+ number=3, 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='max_number_of_parallel_subtasks', full_name='org.apache.flink.fn_execution.v1.UserDefinedDataStreamFunction.RuntimeContext.max_number_of_parallel_subtasks', index=3,
+ number=4, 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='index_of_this_subtask', full_name='org.apache.flink.fn_execution.v1.UserDefinedDataStreamFunction.RuntimeContext.index_of_this_subtask', index=4,
+ number=5, 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='attempt_number', full_name='org.apache.flink.fn_execution.v1.UserDefinedDataStreamFunction.RuntimeContext.attempt_number', index=5,
+ number=6, 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='job_parameters', full_name='org.apache.flink.fn_execution.v1.UserDefinedDataStreamFunction.RuntimeContext.job_parameters', index=6,
+ number=7, type=11, cpp_type=10, label=3,
has_default_value=False, default_value=[],
message_type=None, enum_type=None, containing_type=None,
is_extension=False, extension_scope=None,
options=None, file=DESCRIPTOR),
+ ],
+ extensions=[
+ ],
+ nested_types=[],
+ enum_types=[
+ ],
+ options=None,
+ is_extendable=False,
+ syntax='proto3',
+ extension_ranges=[],
+ oneofs=[
+ ],
+ serialized_start=1177,
+ serialized_end=1480,
+)
+
+_USERDEFINEDDATASTREAMFUNCTION = _descriptor.Descriptor(
+ name='UserDefinedDataStreamFunction',
+ full_name='org.apache.flink.fn_execution.v1.UserDefinedDataStreamFunction',
+ filename=None,
+ file=DESCRIPTOR,
+ containing_type=None,
+ fields=[
_descriptor.FieldDescriptor(
- name='metric_enabled', full_name='org.apache.flink.fn_execution.v1.UserDefinedDataStreamFunctions.metric_enabled', index=1,
- number=2, type=8, cpp_type=7, label=1,
+ name='function_type', full_name='org.apache.flink.fn_execution.v1.UserDefinedDataStreamFunction.function_type', index=0,
+ number=1, type=14, cpp_type=8, 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='runtime_context', full_name='org.apache.flink.fn_execution.v1.UserDefinedDataStreamFunction.runtime_context', index=1,
+ number=2, type=11, cpp_type=10, label=1,
+ has_default_value=False, default_value=None,
+ message_type=None, enum_type=None, containing_type=None,
+ is_extension=False, extension_scope=None,
+ options=None, file=DESCRIPTOR),
+ _descriptor.FieldDescriptor(
+ name='payload', full_name='org.apache.flink.fn_execution.v1.UserDefinedDataStreamFunction.payload', index=2,
+ number=3, type=12, cpp_type=9, label=1,
+ has_default_value=False, default_value=_b(""),
+ message_type=None, enum_type=None, containing_type=None,
+ is_extension=False, extension_scope=None,
+ options=None, file=DESCRIPTOR),
+ _descriptor.FieldDescriptor(
+ name='metric_enabled', full_name='org.apache.flink.fn_execution.v1.UserDefinedDataStreamFunction.metric_enabled', index=3,
+ number=4, 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,
@@ -557,8 +641,9 @@ _USERDEFINEDDATASTREAMFUNCTIONS = _descriptor.Descriptor(
],
extensions=[
],
- nested_types=[],
+ nested_types=[_USERDEFINEDDATASTREAMFUNCTION_JOBPARAMETER, _USERDEFINEDDATASTREAMFUNCTION_RUNTIMECONTEXT, ],
enum_types=[
+ _USERDEFINEDDATASTREAMFUNCTION_FUNCTIONTYPE,
],
options=None,
is_extendable=False,
@@ -566,8 +651,8 @@ _USERDEFINEDDATASTREAMFUNCTIONS = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=1083,
- serialized_end=1218,
+ serialized_start=852,
+ serialized_end=1560,
)
@@ -597,8 +682,8 @@ _USERDEFINEDAGGREGATEFUNCTION_DATAVIEWSPEC_LISTVIEW = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=1683,
- serialized_end=1767,
+ serialized_start=2025,
+ serialized_end=2109,
)
_USERDEFINEDAGGREGATEFUNCTION_DATAVIEWSPEC_MAPVIEW = _descriptor.Descriptor(
@@ -634,8 +719,8 @@ _USERDEFINEDAGGREGATEFUNCTION_DATAVIEWSPEC_MAPVIEW = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=1770,
- serialized_end=1921,
+ serialized_start=2112,
+ serialized_end=2263,
)
_USERDEFINEDAGGREGATEFUNCTION_DATAVIEWSPEC = _descriptor.Descriptor(
@@ -688,8 +773,8 @@ _USERDEFINEDAGGREGATEFUNCTION_DATAVIEWSPEC = _descriptor.Descriptor(
name='data_view', full_name='org.apache.flink.fn_execution.v1.UserDefinedAggregateFunction.DataViewSpec.data_view',
index=0, containing_type=None, fields=[]),
],
- serialized_start=1420,
- serialized_end=1934,
+ serialized_start=1762,
+ serialized_end=2276,
)
_USERDEFINEDAGGREGATEFUNCTION = _descriptor.Descriptor(
@@ -732,8 +817,8 @@ _USERDEFINEDAGGREGATEFUNCTION = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=1221,
- serialized_end=1934,
+ serialized_start=1563,
+ serialized_end=2276,
)
@@ -812,8 +897,8 @@ _USERDEFINEDAGGREGATEFUNCTIONS = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=1937,
- serialized_end=2277,
+ serialized_start=2279,
+ serialized_end=2619,
)
@@ -850,8 +935,8 @@ _SCHEMA_MAPINFO = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=2355,
- serialized_end=2506,
+ serialized_start=2697,
+ serialized_end=2848,
)
_SCHEMA_TIMEINFO = _descriptor.Descriptor(
@@ -880,8 +965,8 @@ _SCHEMA_TIMEINFO = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=2508,
- serialized_end=2537,
+ serialized_start=2850,
+ serialized_end=2879,
)
_SCHEMA_TIMESTAMPINFO = _descriptor.Descriptor(
@@ -910,8 +995,8 @@ _SCHEMA_TIMESTAMPINFO = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=2539,
- serialized_end=2573,
+ serialized_start=2881,
+ serialized_end=2915,
)
_SCHEMA_LOCALZONEDTIMESTAMPINFO = _descriptor.Descriptor(
@@ -940,8 +1025,8 @@ _SCHEMA_LOCALZONEDTIMESTAMPINFO = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=2575,
- serialized_end=2619,
+ serialized_start=2917,
+ serialized_end=2961,
)
_SCHEMA_ZONEDTIMESTAMPINFO = _descriptor.Descriptor(
@@ -970,8 +1055,8 @@ _SCHEMA_ZONEDTIMESTAMPINFO = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=2621,
- serialized_end=2660,
+ serialized_start=2963,
+ serialized_end=3002,
)
_SCHEMA_DECIMALINFO = _descriptor.Descriptor(
@@ -1007,8 +1092,8 @@ _SCHEMA_DECIMALINFO = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=2662,
- serialized_end=2709,
+ serialized_start=3004,
+ serialized_end=3051,
)
_SCHEMA_BINARYINFO = _descriptor.Descriptor(
@@ -1037,8 +1122,8 @@ _SCHEMA_BINARYINFO = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=2711,
- serialized_end=2739,
+ serialized_start=3053,
+ serialized_end=3081,
)
_SCHEMA_VARBINARYINFO = _descriptor.Descriptor(
@@ -1067,8 +1152,8 @@ _SCHEMA_VARBINARYINFO = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=2741,
- serialized_end=2772,
+ serialized_start=3083,
+ serialized_end=3114,
)
_SCHEMA_CHARINFO = _descriptor.Descriptor(
@@ -1097,8 +1182,8 @@ _SCHEMA_CHARINFO = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=2774,
- serialized_end=2800,
+ serialized_start=3116,
+ serialized_end=3142,
)
_SCHEMA_VARCHARINFO = _descriptor.Descriptor(
@@ -1127,8 +1212,8 @@ _SCHEMA_VARCHARINFO = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=2802,
- serialized_end=2831,
+ serialized_start=3144,
+ serialized_end=3173,
)
_SCHEMA_FIELDTYPE = _descriptor.Descriptor(
@@ -1251,8 +1336,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=2834,
- serialized_end=3906,
+ serialized_start=3176,
+ serialized_end=4248,
)
_SCHEMA_FIELD = _descriptor.Descriptor(
@@ -1295,8 +1380,8 @@ _SCHEMA_FIELD = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=3908,
- serialized_end=4016,
+ serialized_start=4250,
+ serialized_end=4358,
)
_SCHEMA = _descriptor.Descriptor(
@@ -1326,8 +1411,8 @@ _SCHEMA = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=2280,
- serialized_end=4308,
+ serialized_start=2622,
+ serialized_end=4650,
)
@@ -1381,8 +1466,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=4389,
- serialized_end=4714,
+ serialized_start=4731,
+ serialized_end=5056,
)
_TYPEINFO_FIELD = _descriptor.Descriptor(
@@ -1425,8 +1510,8 @@ _TYPEINFO_FIELD = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=4716,
- serialized_end=4826,
+ serialized_start=5058,
+ serialized_end=5168,
)
_TYPEINFO = _descriptor.Descriptor(
@@ -1456,8 +1541,8 @@ _TYPEINFO = _descriptor.Descriptor(
extension_ranges=[],
oneofs=[
],
- serialized_start=4311,
- serialized_end=5087,
+ serialized_start=4653,
+ serialized_end=5429,
)
_INPUT.fields_by_name['udf'].message_type = _USERDEFINEDFUNCTION
@@ -1475,9 +1560,12 @@ _USERDEFINEDFUNCTIONS.fields_by_name['udfs'].message_type = _USERDEFINEDFUNCTION
_USERDEFINEDFUNCTIONS.fields_by_name['windows'].message_type = _OVERWINDOW
_OVERWINDOW.fields_by_name['window_type'].enum_type = _OVERWINDOW_WINDOWTYPE
_OVERWINDOW_WINDOWTYPE.containing_type = _OVERWINDOW
-_USERDEFINEDDATASTREAMFUNCTION.fields_by_name['functionType'].enum_type = _USERDEFINEDDATASTREAMFUNCTION_FUNCTIONTYPE
+_USERDEFINEDDATASTREAMFUNCTION_JOBPARAMETER.containing_type = _USERDEFINEDDATASTREAMFUNCTION
+_USERDEFINEDDATASTREAMFUNCTION_RUNTIMECONTEXT.fields_by_name['job_parameters'].message_type = _USERDEFINEDDATASTREAMFUNCTION_JOBPARAMETER
+_USERDEFINEDDATASTREAMFUNCTION_RUNTIMECONTEXT.containing_type = _USERDEFINEDDATASTREAMFUNCTION
+_USERDEFINEDDATASTREAMFUNCTION.fields_by_name['function_type'].enum_type = _USERDEFINEDDATASTREAMFUNCTION_FUNCTIONTYPE
+_USERDEFINEDDATASTREAMFUNCTION.fields_by_name['runtime_context'].message_type = _USERDEFINEDDATASTREAMFUNCTION_RUNTIMECONTEXT
_USERDEFINEDDATASTREAMFUNCTION_FUNCTIONTYPE.containing_type = _USERDEFINEDDATASTREAMFUNCTION
-_USERDEFINEDDATASTREAMFUNCTIONS.fields_by_name['udfs'].message_type = _USERDEFINEDDATASTREAMFUNCTION
_USERDEFINEDAGGREGATEFUNCTION_DATAVIEWSPEC_LISTVIEW.fields_by_name['element_type'].message_type = _SCHEMA_FIELDTYPE
_USERDEFINEDAGGREGATEFUNCTION_DATAVIEWSPEC_LISTVIEW.containing_type = _USERDEFINEDAGGREGATEFUNCTION_DATAVIEWSPEC
_USERDEFINEDAGGREGATEFUNCTION_DATAVIEWSPEC_MAPVIEW.fields_by_name['key_type'].message_type = _SCHEMA_FIELDTYPE
@@ -1585,7 +1673,6 @@ DESCRIPTOR.message_types_by_name['UserDefinedFunction'] = _USERDEFINEDFUNCTION
DESCRIPTOR.message_types_by_name['UserDefinedFunctions'] = _USERDEFINEDFUNCTIONS
DESCRIPTOR.message_types_by_name['OverWindow'] = _OVERWINDOW
DESCRIPTOR.message_types_by_name['UserDefinedDataStreamFunction'] = _USERDEFINEDDATASTREAMFUNCTION
-DESCRIPTOR.message_types_by_name['UserDefinedDataStreamFunctions'] = _USERDEFINEDDATASTREAMFUNCTIONS
DESCRIPTOR.message_types_by_name['UserDefinedAggregateFunction'] = _USERDEFINEDAGGREGATEFUNCTION
DESCRIPTOR.message_types_by_name['UserDefinedAggregateFunctions'] = _USERDEFINEDAGGREGATEFUNCTIONS
DESCRIPTOR.message_types_by_name['Schema'] = _SCHEMA
@@ -1621,18 +1708,27 @@ OverWindow = _reflection.GeneratedProtocolMessageType('OverWindow', (_message.Me
_sym_db.RegisterMessage(OverWindow)
UserDefinedDataStreamFunction = _reflection.GeneratedProtocolMessageType('UserDefinedDataStreamFunction', (_message.Message,), dict(
+
+ JobParameter = _reflection.GeneratedProtocolMessageType('JobParameter', (_message.Message,), dict(
+ DESCRIPTOR = _USERDEFINEDDATASTREAMFUNCTION_JOBPARAMETER,
+ __module__ = 'flink_fn_execution_pb2'
+ # @@protoc_insertion_point(class_scope:org.apache.flink.fn_execution.v1.UserDefinedDataStreamFunction.JobParameter)
+ ))
+ ,
+
+ RuntimeContext = _reflection.GeneratedProtocolMessageType('RuntimeContext', (_message.Message,), dict(
+ DESCRIPTOR = _USERDEFINEDDATASTREAMFUNCTION_RUNTIMECONTEXT,
+ __module__ = 'flink_fn_execution_pb2'
+ # @@protoc_insertion_point(class_scope:org.apache.flink.fn_execution.v1.UserDefinedDataStreamFunction.RuntimeContext)
+ ))
+ ,
DESCRIPTOR = _USERDEFINEDDATASTREAMFUNCTION,
__module__ = 'flink_fn_execution_pb2'
# @@protoc_insertion_point(class_scope:org.apache.flink.fn_execution.v1.UserDefinedDataStreamFunction)
))
_sym_db.RegisterMessage(UserDefinedDataStreamFunction)
-
-UserDefinedDataStreamFunctions = _reflection.GeneratedProtocolMessageType('UserDefinedDataStreamFunctions', (_message.Message,), dict(
- DESCRIPTOR = _USERDEFINEDDATASTREAMFUNCTIONS,
- __module__ = 'flink_fn_execution_pb2'
- # @@protoc_insertion_point(class_scope:org.apache.flink.fn_execution.v1.UserDefinedDataStreamFunctions)
- ))
-_sym_db.RegisterMessage(UserDefinedDataStreamFunctions)
+_sym_db.RegisterMessage(UserDefinedDataStreamFunction.JobParameter)
+_sym_db.RegisterMessage(UserDefinedDataStreamFunction.RuntimeContext)
UserDefinedAggregateFunction = _reflection.GeneratedProtocolMessageType('UserDefinedAggregateFunction', (_message.Message,), dict(
diff --git a/flink-python/pyflink/fn_execution/operation_utils.py b/flink-python/pyflink/fn_execution/operation_utils.py
index 503bddf..c7b51f3 100644
--- a/flink-python/pyflink/fn_execution/operation_utils.py
+++ b/flink-python/pyflink/fn_execution/operation_utils.py
@@ -106,42 +106,41 @@ def _extract_input(args) -> Tuple[str, Dict, List]:
return ",".join(args_str), local_variable_dict, local_funcs
-def extract_data_stream_stateless_funcs(udfs):
+def extract_data_stream_stateless_funcs(udf_proto):
"""
Extracts user-defined-function from the proto representation of a
:class:`Function`.
- :param udfs: the proto representation of the Python
- :class:`Function`
+ :param udf: the proto representation of the Python :class:`Function`
"""
- func_type = udfs[0].functionType
- udf = flink_fn_execution_pb2.UserDefinedDataStreamFunction
+ func_type = udf_proto.function_type
+ UserDefinedDataStreamFunction = flink_fn_execution_pb2.UserDefinedDataStreamFunction
func = None
- if func_type == udf.MAP:
- func = cloudpickle.loads(udfs[0].payload).map
- elif func_type == udf.FLAT_MAP:
- func = cloudpickle.loads(udfs[0].payload).flat_map
- elif func_type == udf.REDUCE:
- reduce_func = cloudpickle.loads(udfs[0].payload).reduce
+ user_defined_func = cloudpickle.loads(udf_proto.payload)
+ if func_type == UserDefinedDataStreamFunction.MAP:
+ func = user_defined_func.map
+ elif func_type == UserDefinedDataStreamFunction.FLAT_MAP:
+ func = user_defined_func.flat_map
+ elif func_type == UserDefinedDataStreamFunction.REDUCE:
+ reduce_func = user_defined_func.reduce
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)
+ elif func_type == UserDefinedDataStreamFunction.CO_MAP:
+ co_map_func = user_defined_func
def wrap_func(value):
return co_map_func.map1(value[1]) if value[0] else co_map_func.map2(value[2])
func = wrap_func
- elif func_type == udf.CO_FLAT_MAP:
- co_flat_map_func = cloudpickle.loads(udfs[0].payload)
+ elif func_type == UserDefinedDataStreamFunction.CO_FLAT_MAP:
+ co_flat_map_func = user_defined_func
def wrap_func(value):
- return co_flat_map_func.flat_map1(
- value[1]) if value[0] else co_flat_map_func.flat_map2(
- value[2])
+ return co_flat_map_func.flat_map1(value[1]) if value[0] else \
+ co_flat_map_func.flat_map2(value[2])
func = wrap_func
- return func
+ return func, user_defined_func
def _parse_constant_value(constant_value) -> Tuple[str, Any]:
diff --git a/flink-python/pyflink/proto/flink-fn-execution.proto b/flink-python/pyflink/proto/flink-fn-execution.proto
index c78bacd..af3d3a5 100644
--- a/flink-python/pyflink/proto/flink-fn-execution.proto
+++ b/flink-python/pyflink/proto/flink-fn-execution.proto
@@ -82,14 +82,26 @@ message UserDefinedDataStreamFunction {
CO_MAP = 3;
CO_FLAT_MAP = 4;
}
- FunctionType functionType = 1;
- bytes payload = 2;
-}
-// A list of user-defined DataStream functions to be executed in a batch.
-message UserDefinedDataStreamFunctions {
- repeated UserDefinedDataStreamFunction udfs = 1;
- bool metric_enabled = 2;
+ message JobParameter {
+ string key = 1;
+ string value = 2;
+ }
+
+ message RuntimeContext {
+ string task_name = 1;
+ string task_name_with_subtasks = 2;
+ int32 number_of_parallel_subtasks = 3;
+ int32 max_number_of_parallel_subtasks = 4;
+ int32 index_of_this_subtask = 5;
+ int32 attempt_number = 6;
+ repeated JobParameter job_parameters = 7;
+ }
+
+ FunctionType function_type = 1;
+ RuntimeContext runtime_context = 2;
+ bytes payload = 3;
+ bool metric_enabled = 4;
}
message UserDefinedAggregateFunction {
diff --git a/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/PythonPartitionCustomOperator.java b/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/PythonPartitionCustomOperator.java
index d7802a2..1c06172 100644
--- a/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/PythonPartitionCustomOperator.java
+++ b/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/PythonPartitionCustomOperator.java
@@ -24,10 +24,14 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.CoreOptions;
import org.apache.flink.python.PythonFunctionRunner;
import org.apache.flink.python.env.PythonEnvironmentManager;
-import org.apache.flink.python.env.beam.ProcessPythonEnvironmentManager;
import org.apache.flink.streaming.api.functions.python.DataStreamPythonFunctionInfo;
import org.apache.flink.streaming.api.runners.python.beam.BeamDataStreamStatelessPythonFunctionRunner;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.apache.flink.streaming.api.utils.PythonOperatorUtils.getUserDefinedDataStreamFunctionProto;
+
/**
* The {@link PythonPartitionCustomOperator} enables us to set the number of partitions for current
* operator dynamically when generating the {@link org.apache.flink.streaming.api.graph.StreamGraph} before executing
@@ -53,18 +57,15 @@ public class PythonPartitionCustomOperator<IN, OUT> extends
@Override
public PythonFunctionRunner createPythonFunctionRunner() throws Exception {
PythonEnvironmentManager pythonEnvironmentManager = createPythonEnvironmentManager();
- if (pythonEnvironmentManager instanceof ProcessPythonEnvironmentManager) {
- ProcessPythonEnvironmentManager envManager = (ProcessPythonEnvironmentManager) pythonEnvironmentManager;
- envManager.setEnvironmentVariable(DATA_STREAM_NUM_PARTITIONS,
- String.valueOf(this.numPartitions));
- }
+ Map<String, String> internalParameters = new HashMap<>();
+ internalParameters.put(DATA_STREAM_NUM_PARTITIONS, String.valueOf(this.numPartitions));
return new BeamDataStreamStatelessPythonFunctionRunner(
getRuntimeContext().getTaskName(),
pythonEnvironmentManager,
inputTypeInfo,
outputTypeInfo,
DATA_STREAM_STATELESS_PYTHON_FUNCTION_URN,
- getUserDefinedDataStreamFunctionsProto(),
+ getUserDefinedDataStreamFunctionProto(pythonFunctionInfo, getRuntimeContext(), internalParameters),
DATA_STREAM_MAP_FUNCTION_CODER_URN,
jobOptions,
getFlinkMetricContainer()
diff --git a/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/PythonReduceOperator.java b/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/PythonReduceOperator.java
index 9f2ba13..d5c0485 100644
--- a/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/PythonReduceOperator.java
+++ b/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/PythonReduceOperator.java
@@ -29,10 +29,14 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.python.PythonFunctionRunner;
import org.apache.flink.streaming.api.functions.python.DataStreamPythonFunctionInfo;
import org.apache.flink.streaming.api.runners.python.beam.BeamDataStreamStatelessPythonFunctionRunner;
-import org.apache.flink.streaming.api.typeutils.PythonTypeUtils;
+import org.apache.flink.streaming.api.utils.PythonTypeUtils;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.types.Row;
+import java.util.Collections;
+
+import static org.apache.flink.streaming.api.utils.PythonOperatorUtils.getUserDefinedDataStreamFunctionProto;
+
/**
* {@link PythonReduceOperator} is responsible for launching beam runner which
* will start a python harness to execute user defined python ReduceFunction.
@@ -116,7 +120,7 @@ public class PythonReduceOperator<OUT>
runnerInputTypeInfo,
outputTypeInfo,
DATA_STREAM_STATELESS_PYTHON_FUNCTION_URN,
- getUserDefinedDataStreamFunctionsProto(),
+ getUserDefinedDataStreamFunctionProto(pythonFunctionInfo, getRuntimeContext(), Collections.EMPTY_MAP),
DATA_STREAM_MAP_FUNCTION_CODER_URN, // reuse map function coder
jobOptions,
getFlinkMetricContainer()
diff --git a/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/StatelessOneInputPythonFunctionOperator.java b/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/StatelessOneInputPythonFunctionOperator.java
index c41cfac..deab827 100644
--- a/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/StatelessOneInputPythonFunctionOperator.java
+++ b/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/StatelessOneInputPythonFunctionOperator.java
@@ -30,15 +30,16 @@ import org.apache.flink.fnexecution.v1.FlinkFnApi;
import org.apache.flink.python.PythonFunctionRunner;
import org.apache.flink.streaming.api.functions.python.DataStreamPythonFunctionInfo;
import org.apache.flink.streaming.api.runners.python.beam.BeamDataStreamStatelessPythonFunctionRunner;
-import org.apache.flink.streaming.api.typeutils.PythonTypeUtils;
+import org.apache.flink.streaming.api.utils.PythonTypeUtils;
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 com.google.protobuf.ByteString;
-
+import java.util.Collections;
import java.util.Map;
+import static org.apache.flink.streaming.api.utils.PythonOperatorUtils.getUserDefinedDataStreamFunctionProto;
+
/**
* {@link StatelessOneInputPythonFunctionOperator} is responsible for launching beam runner which will start a python
* harness to execute user defined python function.
@@ -123,7 +124,7 @@ public class StatelessOneInputPythonFunctionOperator<IN, OUT>
inputTypeInfo,
outputTypeInfo,
DATA_STREAM_STATELESS_PYTHON_FUNCTION_URN,
- getUserDefinedDataStreamFunctionsProto(),
+ getUserDefinedDataStreamFunctionProto(pythonFunctionInfo, getRuntimeContext(), Collections.EMPTY_MAP),
coderUrn,
jobOptions,
getFlinkMetricContainer()
@@ -152,21 +153,4 @@ public class StatelessOneInputPythonFunctionOperator<IN, OUT>
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/StatelessTwoInputPythonFunctionOperator.java b/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/StatelessTwoInputPythonFunctionOperator.java
index 8882e1d..63b8fb6 100644
--- a/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/StatelessTwoInputPythonFunctionOperator.java
+++ b/flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/StatelessTwoInputPythonFunctionOperator.java
@@ -32,16 +32,17 @@ import org.apache.flink.fnexecution.v1.FlinkFnApi;
import org.apache.flink.python.PythonFunctionRunner;
import org.apache.flink.streaming.api.functions.python.DataStreamPythonFunctionInfo;
import org.apache.flink.streaming.api.runners.python.beam.BeamDataStreamStatelessPythonFunctionRunner;
-import org.apache.flink.streaming.api.typeutils.PythonTypeUtils;
+import org.apache.flink.streaming.api.utils.PythonTypeUtils;
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.Collections;
import java.util.Map;
+import static org.apache.flink.streaming.api.utils.PythonOperatorUtils.getUserDefinedDataStreamFunctionProto;
+
/**
* {@link StatelessTwoInputPythonFunctionOperator} is responsible for launching beam
* runner which will start a python harness to execute two-input user defined python function.
@@ -155,7 +156,7 @@ public class StatelessTwoInputPythonFunctionOperator<IN1, IN2, OUT>
runnerInputTypeInfo,
outputTypeInfo,
DATA_STREAM_STATELESS_PYTHON_FUNCTION_URN,
- getUserDefinedDataStreamFunctionsProto(),
+ getUserDefinedDataStreamFunctionProto(pythonFunctionInfo, getRuntimeContext(), Collections.EMPTY_MAP),
coderUrn,
jobOptions,
getFlinkMetricContainer()
@@ -211,21 +212,4 @@ public class StatelessTwoInputPythonFunctionOperator<IN1, IN2, OUT>
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/runners/python/beam/BeamDataStreamStatelessPythonFunctionRunner.java b/flink-python/src/main/java/org/apache/flink/streaming/api/runners/python/beam/BeamDataStreamStatelessPythonFunctionRunner.java
index a1e738e..d79c1eb 100644
--- a/flink-python/src/main/java/org/apache/flink/streaming/api/runners/python/beam/BeamDataStreamStatelessPythonFunctionRunner.java
+++ b/flink-python/src/main/java/org/apache/flink/streaming/api/runners/python/beam/BeamDataStreamStatelessPythonFunctionRunner.java
@@ -23,7 +23,7 @@ import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.fnexecution.v1.FlinkFnApi;
import org.apache.flink.python.env.PythonEnvironmentManager;
import org.apache.flink.python.metric.FlinkMetricContainer;
-import org.apache.flink.streaming.api.typeutils.PythonTypeUtils;
+import org.apache.flink.streaming.api.utils.PythonTypeUtils;
import org.apache.beam.model.pipeline.v1.RunnerApi;
@@ -40,7 +40,7 @@ public class BeamDataStreamStatelessPythonFunctionRunner extends BeamPythonFunct
private final TypeInformation inputType;
private final TypeInformation outputTupe;
- private final FlinkFnApi.UserDefinedDataStreamFunctions userDefinedDataStreamFunctions;
+ private final FlinkFnApi.UserDefinedDataStreamFunction userDefinedDataStreamFunction;
private final String coderUrn;
public BeamDataStreamStatelessPythonFunctionRunner(
@@ -49,20 +49,20 @@ public class BeamDataStreamStatelessPythonFunctionRunner extends BeamPythonFunct
TypeInformation inputType,
TypeInformation outputType,
String functionUrn,
- FlinkFnApi.UserDefinedDataStreamFunctions userDefinedDataStreamFunctions,
+ FlinkFnApi.UserDefinedDataStreamFunction userDefinedDataStreamFunction,
String coderUrn,
Map<String, String> jobOptions,
@Nullable FlinkMetricContainer flinkMetricContainer) {
super(taskName, environmentManager, functionUrn, jobOptions, flinkMetricContainer, null, null);
this.inputType = inputType;
this.outputTupe = outputType;
- this.userDefinedDataStreamFunctions = userDefinedDataStreamFunctions;
+ this.userDefinedDataStreamFunction = userDefinedDataStreamFunction;
this.coderUrn = coderUrn;
}
@Override
protected byte[] getUserDefinedFunctionsProtoBytes() {
- return this.userDefinedDataStreamFunctions.toByteArray();
+ return this.userDefinedDataStreamFunction.toByteArray();
}
@Override
diff --git a/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/utils/PythonOperatorUtils.java b/flink-python/src/main/java/org/apache/flink/streaming/api/utils/PythonOperatorUtils.java
similarity index 65%
rename from flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/utils/PythonOperatorUtils.java
rename to flink-python/src/main/java/org/apache/flink/streaming/api/utils/PythonOperatorUtils.java
index 0c8f974..bdfd708 100644
--- a/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/utils/PythonOperatorUtils.java
+++ b/flink-python/src/main/java/org/apache/flink/streaming/api/utils/PythonOperatorUtils.java
@@ -16,14 +16,19 @@
* limitations under the License.
*/
-package org.apache.flink.table.runtime.operators.python.utils;
+package org.apache.flink.streaming.api.utils;
+import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.fnexecution.v1.FlinkFnApi;
+import org.apache.flink.streaming.api.functions.python.DataStreamPythonFunctionInfo;
import org.apache.flink.table.functions.python.PythonFunctionInfo;
import org.apache.flink.table.planner.typeutils.DataViewUtils;
import com.google.protobuf.ByteString;
+import java.util.Map;
+import java.util.stream.Collectors;
+
import static org.apache.flink.table.runtime.typeutils.PythonTypeUtils.toProtoType;
/**
@@ -51,8 +56,8 @@ public enum PythonOperatorUtils {
}
public static FlinkFnApi.UserDefinedAggregateFunction getUserDefinedAggregateFunctionProto(
- PythonFunctionInfo pythonFunctionInfo,
- DataViewUtils.DataViewSpec[] dataViewSpecs) {
+ PythonFunctionInfo pythonFunctionInfo,
+ DataViewUtils.DataViewSpec[] dataViewSpecs) {
FlinkFnApi.UserDefinedAggregateFunction.Builder builder = FlinkFnApi.UserDefinedAggregateFunction.newBuilder();
builder.setPayload(ByteString.copyFrom(pythonFunctionInfo.getPythonFunction().getSerializedPythonFunction()));
for (Object input : pythonFunctionInfo.getInputs()) {
@@ -89,4 +94,39 @@ public enum PythonOperatorUtils {
}
return builder.build();
}
+
+ public static FlinkFnApi.UserDefinedDataStreamFunction getUserDefinedDataStreamFunctionProto(
+ DataStreamPythonFunctionInfo dataStreamPythonFunctionInfo,
+ RuntimeContext runtimeContext,
+ Map<String, String> internalParameters) {
+ FlinkFnApi.UserDefinedDataStreamFunction.Builder builder =
+ FlinkFnApi.UserDefinedDataStreamFunction.newBuilder();
+ builder.setFunctionType(FlinkFnApi.UserDefinedDataStreamFunction.FunctionType.forNumber(
+ dataStreamPythonFunctionInfo.getFunctionType()));
+ builder.setRuntimeContext(
+ FlinkFnApi.UserDefinedDataStreamFunction.RuntimeContext.newBuilder()
+ .setTaskName(runtimeContext.getTaskName())
+ .setTaskNameWithSubtasks(runtimeContext.getTaskNameWithSubtasks())
+ .setNumberOfParallelSubtasks(runtimeContext.getNumberOfParallelSubtasks())
+ .setMaxNumberOfParallelSubtasks(runtimeContext.getMaxNumberOfParallelSubtasks())
+ .setIndexOfThisSubtask(runtimeContext.getIndexOfThisSubtask())
+ .setAttemptNumber(runtimeContext.getAttemptNumber())
+ .addAllJobParameters(runtimeContext.getExecutionConfig().getGlobalJobParameters()
+ .toMap().entrySet().stream().map(
+ entry -> FlinkFnApi.UserDefinedDataStreamFunction.JobParameter.newBuilder()
+ .setKey(entry.getKey())
+ .setValue(entry.getValue())
+ .build())
+ .collect(Collectors.toList()))
+ .addAllJobParameters(internalParameters.entrySet().stream().map(
+ entry -> FlinkFnApi.UserDefinedDataStreamFunction.JobParameter.newBuilder()
+ .setKey(entry.getKey())
+ .setValue(entry.getValue())
+ .build())
+ .collect(Collectors.toList()))
+ .build());
+ builder.setPayload(ByteString.copyFrom(
+ dataStreamPythonFunctionInfo.getPythonFunction().getSerializedPythonFunction()));
+ return builder.build();
+ }
}
diff --git a/flink-python/src/main/java/org/apache/flink/streaming/api/typeutils/PythonTypeUtils.java b/flink-python/src/main/java/org/apache/flink/streaming/api/utils/PythonTypeUtils.java
similarity index 99%
rename from flink-python/src/main/java/org/apache/flink/streaming/api/typeutils/PythonTypeUtils.java
rename to flink-python/src/main/java/org/apache/flink/streaming/api/utils/PythonTypeUtils.java
index 6de0e97..70a9a30 100644
--- a/flink-python/src/main/java/org/apache/flink/streaming/api/typeutils/PythonTypeUtils.java
+++ b/flink-python/src/main/java/org/apache/flink/streaming/api/utils/PythonTypeUtils.java
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package org.apache.flink.streaming.api.typeutils;
+package org.apache.flink.streaming.api.utils;
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo;
diff --git a/flink-python/src/main/java/org/apache/flink/table/runtime/functions/python/AbstractPythonScalarFunctionFlatMap.java b/flink-python/src/main/java/org/apache/flink/table/runtime/functions/python/AbstractPythonScalarFunctionFlatMap.java
index e861dba..a88b682 100644
--- a/flink-python/src/main/java/org/apache/flink/table/runtime/functions/python/AbstractPythonScalarFunctionFlatMap.java
+++ b/flink-python/src/main/java/org/apache/flink/table/runtime/functions/python/AbstractPythonScalarFunctionFlatMap.java
@@ -24,10 +24,10 @@ import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.fnexecution.v1.FlinkFnApi;
+import org.apache.flink.streaming.api.utils.PythonOperatorUtils;
import org.apache.flink.table.functions.ScalarFunction;
import org.apache.flink.table.functions.python.PythonEnv;
import org.apache.flink.table.functions.python.PythonFunctionInfo;
-import org.apache.flink.table.runtime.operators.python.utils.PythonOperatorUtils;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.table.types.utils.TypeConversions;
import org.apache.flink.types.Row;
diff --git a/flink-python/src/main/java/org/apache/flink/table/runtime/functions/python/PythonTableFunctionFlatMap.java b/flink-python/src/main/java/org/apache/flink/table/runtime/functions/python/PythonTableFunctionFlatMap.java
index 544d07a..2630383 100644
--- a/flink-python/src/main/java/org/apache/flink/table/runtime/functions/python/PythonTableFunctionFlatMap.java
+++ b/flink-python/src/main/java/org/apache/flink/table/runtime/functions/python/PythonTableFunctionFlatMap.java
@@ -25,10 +25,10 @@ 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.fnexecution.v1.FlinkFnApi;
+import org.apache.flink.streaming.api.utils.PythonOperatorUtils;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.table.functions.python.PythonEnv;
import org.apache.flink.table.functions.python.PythonFunctionInfo;
-import org.apache.flink.table.runtime.operators.python.utils.PythonOperatorUtils;
import org.apache.flink.table.runtime.typeutils.PythonTypeUtils;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.table.types.utils.TypeConversions;
diff --git a/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/PythonStreamGroupAggregateOperator.java b/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/PythonStreamGroupAggregateOperator.java
index ab24795..07d1d8e 100644
--- a/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/PythonStreamGroupAggregateOperator.java
+++ b/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/PythonStreamGroupAggregateOperator.java
@@ -39,6 +39,7 @@ import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.operators.InternalTimer;
import org.apache.flink.streaming.api.operators.Triggerable;
import org.apache.flink.streaming.api.operators.python.AbstractOneInputPythonFunctionOperator;
+import org.apache.flink.streaming.api.utils.PythonOperatorUtils;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
@@ -49,7 +50,6 @@ import org.apache.flink.table.planner.plan.utils.KeySelectorUtil;
import org.apache.flink.table.planner.typeutils.DataViewUtils;
import org.apache.flink.table.runtime.functions.CleanupState;
import org.apache.flink.table.runtime.keyselector.RowDataKeySelector;
-import org.apache.flink.table.runtime.operators.python.utils.PythonOperatorUtils;
import org.apache.flink.table.runtime.operators.python.utils.StreamRecordRowDataWrappingCollector;
import org.apache.flink.table.runtime.runners.python.beam.BeamTableStatefulPythonFunctionRunner;
import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
diff --git a/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/arrow/AbstractArrowPythonAggregateFunctionOperator.java b/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/arrow/AbstractArrowPythonAggregateFunctionOperator.java
index c55d3c5..87ae2ec 100644
--- a/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/arrow/AbstractArrowPythonAggregateFunctionOperator.java
+++ b/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/arrow/AbstractArrowPythonAggregateFunctionOperator.java
@@ -40,7 +40,7 @@ import org.apache.flink.table.runtime.operators.python.utils.StreamRecordRowData
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.util.Preconditions;
-import static org.apache.flink.table.runtime.operators.python.utils.PythonOperatorUtils.getUserDefinedFunctionProto;
+import static org.apache.flink.streaming.api.utils.PythonOperatorUtils.getUserDefinedFunctionProto;
/**
* The Abstract class of Arrow Aggregate Operator for Pandas {@link AggregateFunction}.
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 4ed2068..45fd6b7 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
@@ -36,7 +36,7 @@ import java.util.ArrayList;
import java.util.List;
import java.util.ListIterator;
-import static org.apache.flink.table.runtime.operators.python.utils.PythonOperatorUtils.getUserDefinedFunctionProto;
+import static org.apache.flink.streaming.api.utils.PythonOperatorUtils.getUserDefinedFunctionProto;
/**
* The Batch Arrow Python {@link AggregateFunction} Operator for Over Window Aggregation.
diff --git a/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/scalar/AbstractPythonScalarFunctionOperator.java b/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/scalar/AbstractPythonScalarFunctionOperator.java
index d6458bb..6cda4a3 100644
--- a/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/scalar/AbstractPythonScalarFunctionOperator.java
+++ b/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/scalar/AbstractPythonScalarFunctionOperator.java
@@ -21,11 +21,11 @@ package org.apache.flink.table.runtime.operators.python.scalar;
import org.apache.flink.annotation.Internal;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.fnexecution.v1.FlinkFnApi;
+import org.apache.flink.streaming.api.utils.PythonOperatorUtils;
import org.apache.flink.table.functions.ScalarFunction;
import org.apache.flink.table.functions.python.PythonEnv;
import org.apache.flink.table.functions.python.PythonFunctionInfo;
import org.apache.flink.table.runtime.operators.python.AbstractStatelessFunctionOperator;
-import org.apache.flink.table.runtime.operators.python.utils.PythonOperatorUtils;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.util.Preconditions;
diff --git a/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/table/AbstractPythonTableFunctionOperator.java b/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/table/AbstractPythonTableFunctionOperator.java
index e3b1288..8a6b49a 100644
--- a/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/table/AbstractPythonTableFunctionOperator.java
+++ b/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/table/AbstractPythonTableFunctionOperator.java
@@ -21,11 +21,11 @@ package org.apache.flink.table.runtime.operators.python.table;
import org.apache.flink.annotation.Internal;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.fnexecution.v1.FlinkFnApi;
+import org.apache.flink.streaming.api.utils.PythonOperatorUtils;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.table.functions.python.PythonEnv;
import org.apache.flink.table.functions.python.PythonFunctionInfo;
import org.apache.flink.table.runtime.operators.python.AbstractStatelessFunctionOperator;
-import org.apache.flink.table.runtime.operators.python.utils.PythonOperatorUtils;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.util.Preconditions;
diff --git a/flink-python/src/test/java/org/apache/flink/streaming/api/typeutils/PythonTypeUtilsTest.java b/flink-python/src/test/java/org/apache/flink/streaming/api/utils/PythonTypeUtilsTest.java
similarity index 99%
rename from flink-python/src/test/java/org/apache/flink/streaming/api/typeutils/PythonTypeUtilsTest.java
rename to flink-python/src/test/java/org/apache/flink/streaming/api/utils/PythonTypeUtilsTest.java
index 6d193ee..5e12ba2 100644
--- a/flink-python/src/test/java/org/apache/flink/streaming/api/typeutils/PythonTypeUtilsTest.java
+++ b/flink-python/src/test/java/org/apache/flink/streaming/api/utils/PythonTypeUtilsTest.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.flink.streaming.api.typeutils;
+package org.apache.flink.streaming.api.utils;
import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;