Source code for pyflink.common.watermark_strategy

################################################################################
#  Licensed to the Apache Software Foundation (ASF) under one
#  or more contributor license agreements.  See the NOTICE file
#  distributed with this work for additional information
#  regarding copyright ownership.  The ASF licenses this file
#  to you under the Apache License, Version 2.0 (the
#  "License"); you may not use this file except in compliance
#  with the License.  You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS,
#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#  See the License for the specific language governing permissions and
# limitations under the License.
################################################################################
import abc

from typing import Any

from pyflink.common.time import Duration
from pyflink.java_gateway import get_gateway


[docs]class WatermarkStrategy(object): """ The WatermarkStrategy defines how to generate Watermarks in the stream sources. The WatermarkStrategy is a builder/factory for the WatermarkGenerator that generates the watermarks and the TimestampAssigner which assigns the internal timestamp of a record. The convenience methods, for example forBoundedOutOfOrderness(Duration), create a WatermarkStrategy for common built in strategies. """ def __init__(self, j_watermark_strategy): self._j_watermark_strategy = j_watermark_strategy self._timestamp_assigner = None
[docs] def with_timestamp_assigner(self, timestamp_assigner: 'TimestampAssigner') -> \ 'WatermarkStrategy': """ Creates a new WatermarkStrategy that wraps this strategy but instead uses the given a TimestampAssigner by implementing TimestampAssigner interface. :: >>> watermark_strategy = WatermarkStrategy.for_monotonous_timestamps() >>> .with_timestamp_assigner(MyTimestampAssigner()) :param timestamp_assigner: The given TimestampAssigner. :return: A WaterMarkStrategy that wraps a TimestampAssigner. """ self._timestamp_assigner = timestamp_assigner return self
[docs] def with_idleness(self, idle_timeout: Duration) -> 'WatermarkStrategy': """ Creates a new enriched WatermarkStrategy that also does idleness detection in the created WatermarkGenerator. :param idle_timeout: The idle timeout. :return: A new WatermarkStrategy with idle detection configured. """ self._j_watermark_strategy = self._j_watermark_strategy\ .withIdleness(idle_timeout._j_duration) return self
[docs] @staticmethod def for_monotonous_timestamps(): """ Creates a watermark strategy for situations with monotonously ascending timestamps. The watermarks are generated periodically and tightly follow the latest timestamp in the data. The delay introduced by this strategy is mainly the periodic interval in which the watermarks are generated. """ JWaterMarkStrategy = get_gateway().jvm\ .org.apache.flink.api.common.eventtime.WatermarkStrategy return WatermarkStrategy(JWaterMarkStrategy.forMonotonousTimestamps())
[docs] @staticmethod def for_bounded_out_of_orderness(max_out_of_orderness: Duration): """ Creates a watermark strategy for situations where records are out of order, but you can place an upper bound on how far the events are out of order. An out-of-order bound B means that once the an event with timestamp T was encountered, no events older than (T - B) will follow any more. """ JWaterMarkStrategy = get_gateway().jvm \ .org.apache.flink.api.common.eventtime.WatermarkStrategy return WatermarkStrategy( JWaterMarkStrategy.forBoundedOutOfOrderness(max_out_of_orderness._j_duration))
class TimestampAssigner(abc.ABC): """ A TimestampAssigner assigns event time timestamps to elements. These timestamps are used by all functions that operate on event time, for example event time windows. Timestamps can be an arbitrary int value, but all built-in implementations represent it as the milliseconds since the Epoch (midnight, January 1, 1970 UTC), the same way as time.time() does it. """ @abc.abstractmethod def extract_timestamp(self, value: Any, record_timestamp: int) -> int: """ Assigns a timestamp to an element, in milliseconds since the Epoch. This is independent of any particular time zone or calendar. The method is passed the previously assigned timestamp of the element. That previous timestamp may have been assigned from a previous assigner. If the element did not carry a timestamp before, this value is the minimum value of int type. :param value: The element that the timestamp will be assigned to. :param record_timestamp: The current internal timestamp of the element, or a negative value, if no timestamp has been assigned yet. :return: The new timestamp. """ pass