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 2021/07/22 01:07:45 UTC
[flink] 02/03: [FLINK-23401][python] Use ParDoPayload to represent
the user-defined functions 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
commit 9df4b7608aea73d7b2ad3fc6aeee559a84e58c0b
Author: Dian Fu <di...@apache.org>
AuthorDate: Fri Jul 2 16:03:23 2021 +0800
[FLINK-23401][python] Use ParDoPayload to represent the user-defined functions in Python DataStream API
This closes #16541.
---
.../pyflink/fn_execution/beam/beam_operations.py | 36 +++++++++++++---------
.../beam/BeamDataStreamPythonFunctionRunner.java | 24 ++++++++++++---
2 files changed, 40 insertions(+), 20 deletions(-)
diff --git a/flink-python/pyflink/fn_execution/beam/beam_operations.py b/flink-python/pyflink/fn_execution/beam/beam_operations.py
index 6961d7c..04fc5d4 100644
--- a/flink-python/pyflink/fn_execution/beam/beam_operations.py
+++ b/flink-python/pyflink/fn_execution/beam/beam_operations.py
@@ -15,8 +15,10 @@
# See the License for the specific language governing permissions and
# limitations under the License.
################################################################################
-
+from apache_beam.portability import common_urns
+from apache_beam.portability.api import beam_runner_api_pb2
from apache_beam.runners.worker import bundle_processor, operation_specs
+from apache_beam.utils import proto_utils
from pyflink.fn_execution import flink_fn_execution_pb2
from pyflink.fn_execution.coders import from_proto, from_type_info_proto, TimeWindowCoder, \
@@ -113,25 +115,29 @@ def create_pandas_over_window_aggregate_function(
table_operations.PandasBatchOverWindowAggregateFunctionOperation)
-@bundle_processor.BeamTransformFactory.register_urn(
- datastream_operations.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,
- beam_operations.StatelessFunctionOperation,
- datastream_operations.StatelessOperation)
+# ----------------- DataStream --------------------
@bundle_processor.BeamTransformFactory.register_urn(
- datastream_operations.DATA_STREAM_STATEFUL_FUNCTION_URN,
- flink_fn_execution_pb2.UserDefinedDataStreamFunction)
+ common_urns.primitives.PAR_DO.urn, beam_runner_api_pb2.ParDoPayload)
def create_data_stream_keyed_process_function(factory, transform_id, transform_proto, parameter,
consumers):
- return _create_user_defined_function_operation(
- factory, transform_proto, consumers, parameter,
- beam_operations.StatefulFunctionOperation,
- datastream_operations.StatefulOperation)
+ urn = parameter.do_fn.urn
+ payload = proto_utils.parse_Bytes(
+ parameter.do_fn.payload, flink_fn_execution_pb2.UserDefinedDataStreamFunction)
+ if urn == datastream_operations.DATA_STREAM_STATELESS_FUNCTION_URN:
+ return _create_user_defined_function_operation(
+ factory, transform_proto, consumers, payload,
+ beam_operations.StatelessFunctionOperation,
+ datastream_operations.StatelessOperation)
+ else:
+ return _create_user_defined_function_operation(
+ factory, transform_proto, consumers, payload,
+ beam_operations.StatefulFunctionOperation,
+ datastream_operations.StatefulOperation)
+
+
+# ----------------- Utilities --------------------
def _create_user_defined_function_operation(factory, transform_proto, consumers, udfs_proto,
diff --git a/flink-python/src/main/java/org/apache/flink/streaming/api/runners/python/beam/BeamDataStreamPythonFunctionRunner.java b/flink-python/src/main/java/org/apache/flink/streaming/api/runners/python/beam/BeamDataStreamPythonFunctionRunner.java
index a9f5b91..defb696 100644
--- a/flink-python/src/main/java/org/apache/flink/streaming/api/runners/python/beam/BeamDataStreamPythonFunctionRunner.java
+++ b/flink-python/src/main/java/org/apache/flink/streaming/api/runners/python/beam/BeamDataStreamPythonFunctionRunner.java
@@ -28,6 +28,7 @@ import org.apache.flink.runtime.state.KeyedStateBackend;
import org.apache.flink.util.Preconditions;
import org.apache.beam.model.pipeline.v1.RunnerApi;
+import org.apache.beam.runners.core.construction.BeamUrns;
import javax.annotation.Nullable;
@@ -83,11 +84,10 @@ public class BeamDataStreamPythonFunctionRunner extends BeamPythonFunctionRunner
@Override
protected Map<String, RunnerApi.PTransform> getTransforms() {
- return Collections.singletonMap(
- TRANSFORM_ID,
- RunnerApi.PTransform.newBuilder()
- .setUniqueName(TRANSFORM_ID)
- .setSpec(
+ // Use ParDoPayload as a wrapper of the actual payload as timer is only supported in ParDo
+ RunnerApi.ParDoPayload payload =
+ RunnerApi.ParDoPayload.newBuilder()
+ .setDoFn(
RunnerApi.FunctionSpec.newBuilder()
.setUrn(functionUrn)
.setPayload(
@@ -96,6 +96,20 @@ public class BeamDataStreamPythonFunctionRunner extends BeamPythonFunctionRunner
userDefinedDataStreamFunction
.toByteArray()))
.build())
+ .build();
+
+ return Collections.singletonMap(
+ TRANSFORM_ID,
+ RunnerApi.PTransform.newBuilder()
+ .setUniqueName(TRANSFORM_ID)
+ .setSpec(
+ RunnerApi.FunctionSpec.newBuilder()
+ .setUrn(
+ BeamUrns.getUrn(
+ RunnerApi.StandardPTransforms.Primitives
+ .PAR_DO))
+ .setPayload(payload.toByteString())
+ .build())
.putInputs(MAIN_INPUT_NAME, INPUT_COLLECTION_ID)
.putOutputs(MAIN_OUTPUT_NAME, OUTPUT_COLLECTION_ID)
.build());