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 2023/02/15 11:24:31 UTC
[flink] branch master updated: [FLINK-31083][python] Allow OutputTag to be reused
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 29b1be61fef [FLINK-31083][python] Allow OutputTag to be reused
29b1be61fef is described below
commit 29b1be61fefd11bf1ef4c9bed5543c4c6f056f4e
Author: Juntao Hu <ma...@gmail.com>
AuthorDate: Wed Feb 15 18:07:13 2023 +0800
[FLINK-31083][python] Allow OutputTag to be reused
This closes #21941.
---
flink-python/pyflink/datastream/output_tag.py | 23 +++++--
.../pyflink/datastream/tests/test_data_stream.py | 70 ++++++++++++++--------
2 files changed, 64 insertions(+), 29 deletions(-)
diff --git a/flink-python/pyflink/datastream/output_tag.py b/flink-python/pyflink/datastream/output_tag.py
index 78da82d4b4d..6e287dbf95a 100644
--- a/flink-python/pyflink/datastream/output_tag.py
+++ b/flink-python/pyflink/datastream/output_tag.py
@@ -43,6 +43,7 @@ class OutputTag(object):
if not tag_id:
raise ValueError("OutputTag tag_id cannot be None or empty string")
self.tag_id = tag_id
+
if type_info is None:
self.type_info = Types.PICKLED_BYTE_ARRAY()
elif isinstance(type_info, list):
@@ -52,10 +53,22 @@ class OutputTag(object):
else:
self.type_info = type_info
+ self._j_output_tag = None
+
+ def __getstate__(self):
+ # prevent java object to be pickled
+ self.type_info._j_typeinfo = None
+ return self.tag_id, self.type_info
+
+ def __setstate__(self, state):
+ tag_id, type_info = state
+ self.tag_id = tag_id
+ self.type_info = type_info
+
def get_java_output_tag(self):
gateway = get_gateway()
- j_obj = gateway.jvm.org.apache.flink.util.OutputTag(self.tag_id,
- self.type_info.get_java_type_info())
- # deal with serializability
- self.type_info._j_typeinfo = None
- return j_obj
+ if self._j_output_tag is None:
+ self._j_output_tag = gateway.jvm.org.apache.flink.util.OutputTag(
+ self.tag_id, self.type_info.get_java_type_info()
+ )
+ return self._j_output_tag
diff --git a/flink-python/pyflink/datastream/tests/test_data_stream.py b/flink-python/pyflink/datastream/tests/test_data_stream.py
index e136dde7d61..3b7bb626b1a 100644
--- a/flink-python/pyflink/datastream/tests/test_data_stream.py
+++ b/flink-python/pyflink/datastream/tests/test_data_stream.py
@@ -482,6 +482,30 @@ class DataStreamTests(object):
]
self.assert_equals_sorted(expected, self.test_sink.get_results())
+ def test_process_side_output(self):
+ tag = OutputTag("side", Types.INT())
+
+ ds = self.env.from_collection([('a', 0), ('b', 1), ('c', 2)],
+ type_info=Types.ROW([Types.STRING(), Types.INT()]))
+
+ class MyProcessFunction(ProcessFunction):
+
+ def process_element(self, value, ctx: 'ProcessFunction.Context'):
+ yield value[0]
+ yield tag, value[1]
+
+ ds2 = ds.process(MyProcessFunction(), output_type=Types.STRING())
+ main_sink = DataStreamTestSinkFunction()
+ ds2.add_sink(main_sink)
+ side_sink = DataStreamTestSinkFunction()
+ ds2.get_side_output(tag).add_sink(side_sink)
+
+ self.env.execute("test_process_side_output")
+ main_expected = ['a', 'b', 'c']
+ self.assert_equals_sorted(main_expected, main_sink.get_results())
+ side_expected = ['0', '1', '2']
+ self.assert_equals_sorted(side_expected, side_sink.get_results())
+
def test_side_output_chained_with_upstream_operator(self):
tag = OutputTag("side", Types.INT())
@@ -656,6 +680,28 @@ class DataStreamTests(object):
expected = [2, 4, 6]
self.assert_equals_sorted(expected, result)
+ def test_side_output_tag_reusing(self):
+ tag = OutputTag("side", Types.INT())
+
+ class MyProcessFunction(ProcessFunction):
+
+ def process_element(self, value, ctx):
+ yield value
+ yield tag, value * 2
+
+ side1_sink = DataStreamTestSinkFunction()
+ ds = self.env.from_collection([1, 2, 3], Types.INT()).process(MyProcessFunction())
+ ds.get_side_output(tag).add_sink(side1_sink)
+
+ side2_sink = DataStreamTestSinkFunction()
+ ds.map(lambda i: i*2).process(MyProcessFunction()).get_side_output(tag).add_sink(side2_sink)
+
+ self.env.execute("test_side_output_tag_reusing")
+ result1 = [i for i in side1_sink.get_results(stringify=False)]
+ result2 = [i for i in side2_sink.get_results(stringify=False)]
+ self.assert_equals_sorted(['2', '4', '6'], result1)
+ self.assert_equals_sorted(['4', '8', '12'], result2)
+
class DataStreamStreamingTests(DataStreamTests):
@@ -996,30 +1042,6 @@ class ProcessDataStreamTests(DataStreamTests):
"current_value: Row(f0=4, f1='1603708289000')"]
self.assert_equals_sorted(expected, results)
- def test_process_side_output(self):
- tag = OutputTag("side", Types.INT())
-
- ds = self.env.from_collection([('a', 0), ('b', 1), ('c', 2)],
- type_info=Types.ROW([Types.STRING(), Types.INT()]))
-
- class MyProcessFunction(ProcessFunction):
-
- def process_element(self, value, ctx: 'ProcessFunction.Context'):
- yield value[0]
- yield tag, value[1]
-
- ds2 = ds.process(MyProcessFunction(), output_type=Types.STRING())
- main_sink = DataStreamTestSinkFunction()
- ds2.add_sink(main_sink)
- side_sink = DataStreamTestSinkFunction()
- ds2.get_side_output(tag).add_sink(side_sink)
-
- self.env.execute("test_process_side_output")
- main_expected = ['a', 'b', 'c']
- self.assert_equals_sorted(main_expected, main_sink.get_results())
- side_expected = ['0', '1', '2']
- self.assert_equals_sorted(side_expected, side_sink.get_results())
-
class ProcessDataStreamStreamingTests(DataStreamStreamingTests, ProcessDataStreamTests,
PyFlinkStreamingTestCase):