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 85532401e2d [FLINK-40829][python] Allow assign uid for Timestamp
operators (#29312)
85532401e2d is described below
commit 85532401e2da6c7b19e67c32b02f8f2791120e52
Author: florianvazelle <[email protected]>
AuthorDate: Wed Oct 21 11:30:52 2026 +0200
[FLINK-40829][python] Allow assign uid for Timestamp operators (#29312)
---
flink-python/pyflink/datastream/data_stream.py | 68 +++++++++++++++++++++-
.../pyflink/datastream/tests/test_data_stream.py | 60 +++++++++++++++++++
2 files changed, 126 insertions(+), 2 deletions(-)
diff --git a/flink-python/pyflink/datastream/data_stream.py
b/flink-python/pyflink/datastream/data_stream.py
index 8be48e0640f..14d660abd45 100644
--- a/flink-python/pyflink/datastream/data_stream.py
+++ b/flink-python/pyflink/datastream/data_stream.py
@@ -15,6 +15,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
################################################################################
+import hashlib
import typing
import uuid
from enum import Enum
@@ -65,6 +66,19 @@ __all__ = ['CloseableIterator', 'DataStream', 'KeyedStream',
'ConnectedStreams',
WINDOW_STATE_NAME = 'window-contents'
+# Appended to the uid of an assign_timestamps_and_watermarks() stream, to
build one for each of
+# the operators it hides. They are part of the uid, so they must stay as they
are.
+UID_SUFFIX_EXTRACT_TIMESTAMP = 'extract-timestamp'
+UID_SUFFIX_TIMESTAMPS_AND_WATERMARKS = 'timestamps-and-watermarks'
+
+
+def _derive_uid_hash(uid_hash: str, suffix: str) -> str:
+ """
+ Derives the hash of an operator the user cannot reach from the hash of one
they can. A hash
+ must be 32 hex digits, so the suffix cannot be appended to it as it is to
a uid.
+ """
+ return hashlib.sha256(("%s-%s" % (uid_hash,
suffix)).encode('utf-8')).hexdigest()[:32]
+
class DataStream(object):
"""
@@ -704,8 +718,13 @@ class DataStream(object):
# step 3: remove the timestamp field which is added in step 1
JRemoveTimestampMapFunction =
gateway.jvm.org.apache.flink.streaming.api.functions \
.python.eventtime.RemoveTimestampMapFunction
- result = DataStream(j_watermarked_data_stream.map(
- JRemoveTimestampMapFunction(), self._j_data_stream.getType()))
+ result = _TimestampsAndWatermarksDataStream(
+ j_watermarked_data_stream.map(
+ JRemoveTimestampMapFunction(),
self._j_data_stream.getType()),
+ [(UID_SUFFIX_EXTRACT_TIMESTAMP,
+ timestamped_data_stream._j_data_stream.getTransformation()),
+ (UID_SUFFIX_TIMESTAMPS_AND_WATERMARKS,
+ j_watermarked_data_stream.getTransformation())])
result.name("Remove-Timestamp")
return result
else:
@@ -963,6 +982,51 @@ class DataStream(object):
return self
+class _TimestampsAndWatermarksDataStream(DataStream):
+ """
+ The stream which :func:`DataStream.assign_timestamps_and_watermarks`
returns for a Python
+ TimestampAssigner.
+
+ Such an assigner needs three operators: one extracts the timestamp, one
assigns the
+ timestamps and the watermarks, and one removes the extracted timestamp
again. Only the last
+ one is represented by the returned stream, so a uid or a uid hash set on
it would leave the
+ two others without one. They then make the job impossible to submit if
auto generated uids
+ are disabled, see
:func:`~pyflink.common.ExecutionConfig.disable_auto_generated_uids`. This
+ stream gives them one too, derived from the one it receives.
+ """
+
+ def __init__(self, j_data_stream, hidden_j_transformations):
+ super(_TimestampsAndWatermarksDataStream, self).__init__(j_data_stream)
+ # The (uid suffix, transformation) pairs of the two operators the
assigner hides.
+ self._hidden_j_transformations = hidden_j_transformations
+
+ def uid(self, uid: str) -> 'DataStream':
+ """
+ Sets an ID for this operator, and one derived from it for each of the
operators the
+ timestamp assigner hides.
+
+ :param uid: The unique user-specified ID of this transformation.
+ :return: The operator with the specified ID.
+ """
+ super(_TimestampsAndWatermarksDataStream, self).uid(uid)
+ for uid_suffix, j_transformation in self._hidden_j_transformations:
+ j_transformation.setUid("%s-%s" % (uid, uid_suffix))
+ return self
+
+ def set_uid_hash(self, uid_hash: str) -> 'DataStream':
+ """
+ Sets an user provided hash for this operator, and one derived from it
for each of the
+ operators the timestamp assigner hides.
+
+ :param uid_hash: The user provided hash for this operator.
+ :return: The operator with the user provided hash.
+ """
+ super(_TimestampsAndWatermarksDataStream, self).set_uid_hash(uid_hash)
+ for uid_suffix, j_transformation in self._hidden_j_transformations:
+ j_transformation.setUidHash(_derive_uid_hash(uid_hash, uid_suffix))
+ return self
+
+
class DataStreamSink(object):
"""
A Stream Sink. This is used for emitting elements from a streaming
topology.
diff --git a/flink-python/pyflink/datastream/tests/test_data_stream.py
b/flink-python/pyflink/datastream/tests/test_data_stream.py
index 272473b8450..da91445914e 100644
--- a/flink-python/pyflink/datastream/tests/test_data_stream.py
+++ b/flink-python/pyflink/datastream/tests/test_data_stream.py
@@ -1537,6 +1537,66 @@ class CommonDataStreamTests(PyFlinkTestCase):
pre_ship_strategy = shuffle_node['predecessors'][0]['ship_strategy']
self.assertEqual(pre_ship_strategy, 'SHUFFLE')
+ def _assign_timestamps(self):
+ class MyTimestampAssigner(TimestampAssigner):
+
+ def extract_timestamp(self, value, record_timestamp) -> int:
+ return int(value[0])
+
+ self.env.get_config().disable_auto_generated_uids()
+
+ ds = self.env.from_collection([(1603708211000, 'hi'), (1603708224000,
'hello')],
+ type_info=Types.TUPLE([Types.LONG(),
Types.STRING()]))
+ return ds.uid('source') \
+ .assign_timestamps_and_watermarks(
+ WatermarkStrategy.for_monotonous_timestamps()
+ .with_timestamp_assigner(MyTimestampAssigner()))
+
+ def _uids_and_hashes(self, name):
+ """The uid and the uid hash of each operator with the given name, in
creation order."""
+ j_transformations = {}
+ for j_sink in
self.env._j_stream_execution_environment.getTransformations():
+ for j_transformation in j_sink.getTransitivePredecessors():
+ j_transformations[j_transformation.getId()] = j_transformation
+ return [(j_transformation.getUid(),
j_transformation.getUserProvidedNodeHash())
+ for _, j_transformation in sorted(j_transformations.items())
+ if j_transformation.getName() == name]
+
+ def test_timestamp_assigner_uid(self):
+ """A uid on the stream must cover the operators the timestamp assigner
hides."""
+
self._assign_timestamps().uid('watermarks').add_sink(self.test_sink).uid('sink')
+
+ # Without a uid on every operator, this raises an
IllegalStateException.
+ self.env.get_execution_plan()
+
+ # The uids identify the operators in a savepoint, so they must not
change.
+ self.assertEqual(self._uids_and_hashes('Extract-Timestamp'),
+ [('watermarks-extract-timestamp', None)])
+ self.assertEqual(self._uids_and_hashes('Timestamps/Watermarks'),
+ [('watermarks-timestamps-and-watermarks', None)])
+ self.assertEqual(self._uids_and_hashes('Remove-Timestamp'),
[('watermarks', None)])
+
+ def test_timestamp_assigner_uid_hash(self):
+
self._assign_timestamps().set_uid_hash('0123456789abcdef0123456789abcdef') \
+ .add_sink(self.test_sink).uid('sink')
+
+ self.env.get_execution_plan()
+
+ self.assertEqual(self._uids_and_hashes('Extract-Timestamp'),
+ [(None, '2230997a8b7cde93ec413c77bd3b152a')])
+ self.assertEqual(self._uids_and_hashes('Timestamps/Watermarks'),
+ [(None, 'f20476c038a8d6395d90768849f9d94b')])
+
+ def test_timestamp_assigner_uid_with_python_operators(self):
+ self._assign_timestamps().uid('watermarks') \
+ .map(lambda x: x, output_type=Types.TUPLE([Types.LONG(),
Types.STRING()])) \
+ .uid('map') \
+ .key_by(lambda x: x[1], key_type=Types.STRING()) \
+ .reduce(lambda a, b: (a[0] + b[0], a[1])).uid('reduce') \
+ .add_sink(self.test_sink).uid('sink')
+
+ self.env.get_execution_plan()
+
def test_keyed_stream_partitioning(self):
ds = self.env.from_collection([('ab', 1), ('bdc', 2), ('cfgs', 3),
('deeefg', 4)])
keyed_stream = ds.key_by(lambda x: x[1])
