Watermark

Watermark это метка времени, которую система потоковой обработки продвигает вперёд по мере поступления событий и использует как сигнал о том, что данные до этой отметки, скорее всего, уже пришли. Термин используют и Apache Flink, и Apache Beam, у обоих он отвечает на один и тот же практический вопрос, сколько ещё ждать перед тем, как закрыть окно и выдать результат агрегации по моменту, когда событие реально произошло (event time), а не по моменту, когда его обработала система (processing time).

Без watermark обработка по event time вообще не имеет смысла. Система может либо ждать данные бесконечно, либо закрывать окна по своим внутренним часам и терять всё, что пришло с сетевой задержкой или из офлайн-источника. Watermark даёт третий вариант, управляемый компромисс между задержкой ответа и полнотой данных.

Что такое Watermark в потоковой обработке данных

Официальная документация Apache Flink формулирует это так, watermark сообщает системе о прогрессе по event time. Apache Beam говорит про то же самое чуть шире и определяет watermark как понятие полноты входных данных (notion of input completeness), способ системы понять, что для окна, вероятно, уже пришли все данные. Формулировки различаются терминологией, но по сути описывают одно и то же явление, сравнение вынесено в таблицу в разделе про Apache Beam ниже. Программный гайд Beam прямо связывает watermark с неограниченной коллекцией (unbounded PCollection), потоком без заранее известного конца.

Практическая причина, почему нельзя просто закрывать окно по системным часам, в том, что событие может прийти позже, чем произошло. Мобильный клиент теряет связь и досылает данные через минуту, один источник в кластере отстаёт от другого по сетевой задержке, партиция Kafka читается с опозданием из-за перебалансировки. Watermark абстрагирует эту неопределённость в одно число, ниже которого можно с приемлемой вероятностью считать, что новых данных уже не будет.

Архитектура watermark в Apache Flink

По документации Flink о генерации watermark, механизм строится на трёх компонентах.

• TimestampAssigner. Извлекает или назначает событию timestamp, то есть отвечает на вопрос, когда это событие произошло на самом деле, а не когда система его увидела.
• WatermarkGenerator. Генерирует watermark на основе уже увиденных событий, встроенная стратегия bounded out-of-orderness делает это как максимальный увиденный timestamp минус фиксированная задержка на возможный беспорядок в приходе данных.
• WatermarkStrategy. Объединяет TimestampAssigner и WatermarkGenerator в единый интерфейс и передаётся источнику или оператору целиком, дополнительно к нему подключается пометка простаивающих источников и выравнивание нескольких потоков.

Все три компонента передаются источнику или оператору вместе, отдельно ни TimestampAssigner, ни WatermarkGenerator не работают. Курс по потоковой обработке данных на Apache Flink разбирает эти три компонента и оконный API на практике, если нужен переход от концепции к продакшену.

Принцип работы watermark

Оператор обязан полностью обработать пришедший watermark, прежде чем передать его дальше по графу вычислений, поэтому продвижение watermark идёт волной от источника к приёмнику, а не мгновенно по всему пайплайну сразу.

Периодическая и punctuated генерация

Документация Flink различает два способа эмиссии watermark. При периодической генерации фреймворк с заданным интервалом вызывает у генератора метод onPeriodicEmit(), который решает, стоит ли продвинуть watermark дальше, именно так работает встроенная стратегия bounded out-of-orderness. При punctuated генерации watermark эмитится не по расписанию, а сразу при обнаружении в потоке специального маркерного события, интервал ожидания не нужен вовсе, зато нужен источник, который такие маркеры действительно присылает.

Периодический вариант не требует контроля над форматом источника, поэтому демо ниже строится именно на нём, punctuated генерации нужен источник, который сам присылает маркерные события, а Kafka-топик из этой статьи такого не делает.

Watermark на операторах с несколькими входами

Оператор join, union или любой другой с несколькими входами получает watermark по каждому входу отдельно и продвигает свой собственный watermark как минимум по всем входам. Следствие прямое, один медленный или простаивающий источник тормозит весь оператор целиком, пока не подаст признаков жизни, потому что итоговый watermark не может обогнать самый отстающий вход.

Тот же эффект возникает и внутри одного топика Kafka с несколькими партициями, каждая партиция для Flink это отдельный вход со своим watermark, и застрявшая партиция блокирует прогресс остальных ровно тем же способом, что и застрявший join.

Watermark alignment и idle-источники

Простаивающий источник, партиция без новых сообщений, решается пометкой idle. Источник, помеченный как простаивающий, перестаёт учитываться при вычислении общего watermark как минимума по входам, и остальные потоки продолжают продвигаться без оглядки на него. В PyFlink это делает метод with_idleness у WatermarkStrategy с таймаутом простоя.

Обратная проблема называется watermark alignment. Если один источник обрабатывает данные заметно быстрее остальных, он уходит вперёд и заставляет downstream-операторы буферизовать избыточные данные в ожидании более медленных входов, растёт потребление памяти state backend. Выравнивание с настраиваемым порогом дрейфа не даёт быстрым источникам уходить далеко вперёд, документация описывает порог как configurable drift threshold без конкретных значений по умолчанию, они зависят от нагрузки и подбираются под конкретный пайплайн. Практический разбор тюнинга watermark именно с источником на Kafka есть в статье блога школы «Водяные знаки в заданиях Flink для потоковой обработки данных из Kafka», она куда подробнее разбирает настройку, чем формат энциклопедической статьи позволяет здесь.

Watermark в Apache Beam

Beam описывает тот же механизм в терминах heuristic (эвристического) и perfect (точного) watermark. Perfect watermark гарантированно не пропускает ни одного события старше своего значения, но программный гайд указывает буквально, что «perfect watermarks may not be achievable in all streaming sources», то есть такая гарантия доступна не для всех потоковых источников, для многих реальных систем достижим только heuristic watermark, оценка по наблюдаемым паттернам без строгой гарантии.

Ответственность за отслеживание и продвижение watermark в Beam лежит на раннере (runner), а не на самом SDK, поэтому конкретная реализация отличается от движка к движку, который исполняет Beam-пайплайн. Watermark вместе с allowed lateness определяет здесь, когда триггеры считают данные опоздавшими и что с ними делать дальше, тот же принцип, что у allowedLateness в Flink, только терминология про триггеры своя.

Сценарии использования и ограничения

Watermark нужен там, где обработка идёт по event time, а не по processing time, то есть где порядок и момент прихода данных в систему не совпадают с порядком и моментом их фактического возникновения.

• Мобильные и IoT-клиенты. Устройство теряет связь и досылает накопленные события пачкой уже после восстановления сети, порядок прихода не совпадает с порядком возникновения.
• Несколько источников с разной задержкой. Один сервис пишет в топик быстрее другого, и агрегация, которая ждёт оба потока, обязана ориентироваться на самый медленный.
• Восстановление после сбоя. После перезапуска источник вычитывает накопившийся backlog рывком, и без idle-пометки соседних источников общий watermark на это время замирает.

Общий признак всех трёх сценариев один, порядок прихода событий в систему нельзя доверять как порядку их возникновения. У механизма есть и обратная сторона. Числовые лимиты, максимальная разумная задержка bounded out-of-orderness, объём state backend при большом allowedLateness, пороги watermark alignment, документация не фиксирует как универсальные константы, потому что они целиком зависят от профиля конкретного потока данных и подбираются экспериментально. Явных указаний «когда watermark не стоит применять» ни в документации Flink, ни в программном гайде Beam нет, обе страницы описывают, как включать и настраивать watermark, а не когда от него отказываться.

Практика watermark на PyFlink и Kafka

Демо на PyFlink 2.3.0 (DataStream API) поднимает собственный одноброкерный Kafka в режиме KRaft и показывает генерацию watermark и обработку опоздавших событий на живом, реально работающем во времени источнике, а не на заранее подготовленном датасете. По традиции весь код используемый в статье выкладываем на наш GitHub репозиторий.

Первый скрипт, watermark_generation.py, собирает WatermarkStrategy из TimestampAssigner (извлекает event time из JSON-поля события), встроенного генератора bounded out-of-orderness с задержкой в секунду и with_idleness с таймаутом в 3 секунды. Топик watermark_events создан на 2 партиции нарочно, датчик sensor-1 шлёт события на всём протяжении прогона, а sensor-2 замолкает после третьего сообщения, имитируя простаивающий источник из раздела про idle-источники выше.

# PyFlink (apache-flink) 2.3.0, apache-beam 2.61.0, py4j 0.10.9.7, JDK 17.0.19 (Homebrew
# openjdk@17), коннектор flink-sql-connector-kafka-5.0.0-2.2.jar (собран для Flink 2.2, но
# рабочий и на Flink 2.3.0 - совместимость подтверждена прогоном на стенде), брокер
# apache/kafka:4.3.0 (KRaft), прогнано на стенде 2026-09-05.
#
# Демонстрирует три компонента из документации Flink (generating_watermarks):
# TimestampAssigner - достаёт event time из полезной нагрузки Kafka-сообщения;
# WatermarkGenerator - встроенная периодическая стратегия bounded-out-of-orderness;
# WatermarkStrategy - связывает первые два и добавляет idle-detection.
#
# Топик watermark_events создан с 2 партициями (см. RUNBOOK). Ключ Kafka-сообщения - это
# sensor_id, поэтому каждый датчик попадает в одну и ту же партицию стабильно. sensor-2
# в этом сценарии перестаёт слать события на середине прогона - его партиция замолкает.
# Без пометки простаивающего источника (with_idleness) общий watermark равен минимуму по
# всем партициям (см. документацию Flink и research-файл, п.3) и навсегда застревает на
# минимальном значении, потому что молчащая партиция никогда не продвигает свой watermark
# дальше. Проверено прогоном без with_idleness на этом стенде: ни один таймер и ни одно
# окно не сработали за 12+ секунд, хотя sensor-1 присылал события всё это время. Здесь
# with_idleness включён, чтобы демонстрировать рабочую, а не сломанную конфигурацию.
import json
import threading
import time

from pyflink.common import Duration, Time, WatermarkStrategy
from pyflink.common.serialization import SimpleStringSchema
from pyflink.common.watermark_strategy import TimestampAssigner
from pyflink.datastream import ProcessWindowFunction, StreamExecutionEnvironment
from pyflink.datastream.connectors.kafka import KafkaOffsetsInitializer, KafkaSource
from pyflink.datastream.window import TumblingEventTimeWindows

BOOTSTRAP_SERVERS = "localhost:9094"
TOPIC = "watermark_events"
KAFKA_CONNECTOR_JAR = (
    "file://" + __file__.rsplit("/", 1)[0] + "/jars/flink-sql-connector-kafka-5.0.0-2.2.jar"
)
WINDOW_SECONDS = 4
OUT_OF_ORDERNESS_MS = 1000
IDLE_TIMEOUT_S = 3

START = time.time()


def log(msg: str) -> None:
    print(f"[{time.time() - START:6.1f}s] {msg}", flush=True)


class ExtractEventTime(TimestampAssigner):
    """K3 архитектуры: TimestampAssigner. Достаёт event time из JSON-поля 'ts' (мс)."""

    def extract_timestamp(self, value, record_timestamp):
        return json.loads(value)["ts"]


class PrintWindowResult(ProcessWindowFunction):
    """Печатает границы окна, watermark в момент срабатывания и агрегат по датчику."""

    def process(self, key, context, elements):
        rows = [json.loads(e) for e in elements]
        window = context.window()
        values = [r["value"] for r in rows]
        log(
            f"WINDOW FIRED sensor={key} range=[{window.start},{window.end}) "
            f"watermark={context.current_watermark()} n={len(values)} "
            f"avg={sum(values) / len(values):.2f}"
        )
        yield (key, window.start, window.end, len(values), sum(values) / len(values))


def produce_events():
    """Отдельный поток-продюсер: sensor-1 шлёт события всю дорогу, sensor-2 замолкает
    после трёх сообщений - это и есть простаивающий источник, который проверяет with_idleness."""
    time.sleep(2)  # дать Flink-джобе время подписаться (starting_offsets=latest)
    from confluent_kafka import Producer

    producer = Producer({"bootstrap.servers": BOOTSTRAP_SERVERS})
    base = int(time.time() * 1000)

    def send(sensor, offset_ms, value):
        ts = base + offset_ms
        payload = json.dumps({"sensor": sensor, "ts": ts, "value": value}).encode()
        producer.produce(TOPIC, key=sensor.encode(), value=payload)
        producer.flush(5)
        log(f"produced sensor={sensor} ts_offset={offset_ms} value={value}")

    for i in range(11):
        send("sensor-1", i * 1000, float(20 + i))
        if i < 3:
            send("sensor-2", i * 1000, float(100 + i))
        time.sleep(1.0)
    log("producer done")


def main():
    env = StreamExecutionEnvironment.get_execution_environment()
    env.add_jars(KAFKA_CONNECTOR_JAR)
    env.set_parallelism(1)

    source = (
        KafkaSource.builder()
        .set_bootstrap_servers(BOOTSTRAP_SERVERS)
        .set_topics(TOPIC)
        .set_group_id("watermark_generation_demo")
        .set_starting_offsets(KafkaOffsetsInitializer.latest())
        .set_value_only_deserializer(SimpleStringSchema())
        .build()
    )

    strategy = (
        WatermarkStrategy.for_bounded_out_of_orderness(Duration.of_millis(OUT_OF_ORDERNESS_MS))
        .with_timestamp_assigner(ExtractEventTime())
        .with_idleness(Duration.of_seconds(IDLE_TIMEOUT_S))
    )

    stream = env.from_source(source, strategy, "kafka-watermark-events")
    (
        stream.key_by(lambda v: json.loads(v)["sensor"])
        .window(TumblingEventTimeWindows.of(Time.seconds(WINDOW_SECONDS)))
        .process(PrintWindowResult())
    )

    producer_thread = threading.Thread(target=produce_events, daemon=True)
    producer_thread.start()

    job_client = env.execute_async("watermark generation demo")
    producer_thread.join()
    time.sleep(6)  # дать последним окнам сработать после того, как watermark догонит их
    job_client.cancel()
    log("job cancelled, demo finished")


if __name__ == "__main__":
    main()

Реальный вывод прогона показывает главное, окно датчика sensor-2 закрывается вовремя вместе с окном sensor-1, хотя sensor-2 к этому моменту молчит уже несколько секунд.

[  10.4s] WINDOW FIRED sensor=sensor-1 range=[1788626576000,1788626580000) watermark=1788626583610 n=1 avg=20.00
[  10.4s] WINDOW FIRED sensor=sensor-2 range=[1788626576000,1788626580000) watermark=1788626583610 n=1 avg=100.00
[  12.6s] WINDOW FIRED sensor=sensor-2 range=[1788626580000,1788626584000) watermark=1788626585627 n=2 avg=101.50
[  12.6s] WINDOW FIRED sensor=sensor-1 range=[1788626580000,1788626584000) watermark=1788626585627 n=4 avg=22.50

Без with_idleness та же джоба на том же стенде не закрыла ни одного окна за 21 секунду прогона, хотя sensor-1 всё это время исправно присылал события, ровно та блокировка минимумом по входам, которая описана в разделе про watermark alignment и idle-источники.

Второй скрипт, late_event_handling.py, разбирает allowedLateness и side output для событий, которые пришли позже, чем watermark уже прошёл конец их окна. В пределах allowedLateness окно обязано пересчитаться заново, late firing, а за его пределами событие уходит в отдельный поток через OutputTag вместо того, чтобы молча потеряться.

# PyFlink (apache-flink) 2.3.0, apache-beam 2.61.0, py4j 0.10.9.7, JDK 17.0.19 (Homebrew
# openjdk@17), коннектор flink-sql-connector-kafka-5.0.0-2.2.jar (собран для Flink 2.2, но
# рабочий и на Flink 2.3.0 - совместимость подтверждена прогоном на стенде), брокер
# apache/kafka:4.3.0 (KRaft), прогнано на стенде 2026-09-05.
#
# Демонстрирует allowedLateness и side output для событий, которые пришли позже, чем
# watermark уже продвинулся мимо конца их окна:
# - в пределах allowedLateness окно пересчитывается повторно (late firing);
# - за пределами allowedLateness событие уходит в side output через OutputTag, а не
#   теряется молча.
#
# "sensor-1" - единственный ключ, чьё окно мы наблюдаем. Второй ключ "clock" нужен только
# чтобы толкать общий watermark вперёд, не касаясь состояния окон "sensor-1": на этом
# стенде подтверждено прогоном, что искусственное "далёкое вперёд" событие ПОД ТЕМ ЖЕ
# ключом, что и наблюдаемое окно, иногда попадает в чужой elements при печати внутри
# ProcessWindowFunction, когда для одного ключа одновременно открыто больше одного окна -
# со своим ключом "clock" этого не происходит ни разу за серию прогонов. Это граница
# метода наблюдения, а не свойство watermark как такового.
import json
import threading
import time

from pyflink.common import Duration, Time, WatermarkStrategy
from pyflink.common.serialization import SimpleStringSchema
from pyflink.common.typeinfo import Types
from pyflink.common.watermark_strategy import TimestampAssigner
from pyflink.datastream import ProcessFunction, ProcessWindowFunction, StreamExecutionEnvironment
from pyflink.datastream.connectors.kafka import KafkaOffsetsInitializer, KafkaSource
from pyflink.datastream.output_tag import OutputTag
from pyflink.datastream.window import TumblingEventTimeWindows

BOOTSTRAP_SERVERS = "localhost:9094"
TOPIC = "watermark_events"
KAFKA_CONNECTOR_JAR = (
    "file://" + __file__.rsplit("/", 1)[0] + "/jars/flink-sql-connector-kafka-5.0.0-2.2.jar"
)
WINDOW_SECONDS = 5
ALLOWED_LATENESS_MS = 4000
OUT_OF_ORDERNESS_MS = 500

START = time.time()
LATE_DATA_TAG = OutputTag("late-data", Types.STRING())


def log(msg: str) -> None:
    print(f"[{time.time() - START:6.1f}s] {msg}", flush=True)


class ExtractEventTime(TimestampAssigner):
    def extract_timestamp(self, value, record_timestamp):
        return json.loads(value)["ts"]


class PrintWindowResult(ProcessWindowFunction):
    """Основной выход окна. Late firing виден здесь же: то же окно печатается повторно,
    с бОльшим n и более поздним watermark, а не как отдельное событие где-то ещё."""

    def process(self, key, context, elements):
        rows = [json.loads(e) for e in elements]
        window = context.window()
        offsets = sorted(r["off"] for r in rows)
        log(
            f"WINDOW key={key} range=[{window.start},{window.end}) "
            f"watermark={context.current_watermark()} n={len(rows)} offsets={offsets}"
        )
        yield (key, window.start, window.end, len(rows))


class PrintLateData(ProcessFunction):
    """Side output: события, которые пришли позже window_end + allowedLateness."""

    def process_element(self, value, ctx):
        row = json.loads(value)
        log(
            f"SIDE OUTPUT (слишком поздно) key={row['sensor']} off={row['off']} "
            f"watermark={ctx.timer_service().current_watermark()}"
        )
        yield value


def build_job(group_id: str):
    env = StreamExecutionEnvironment.get_execution_environment()
    env.add_jars(KAFKA_CONNECTOR_JAR)
    env.set_parallelism(1)

    source = (
        KafkaSource.builder()
        .set_bootstrap_servers(BOOTSTRAP_SERVERS)
        .set_topics(TOPIC)
        .set_group_id(group_id)
        .set_starting_offsets(KafkaOffsetsInitializer.latest())
        .set_value_only_deserializer(SimpleStringSchema())
        .build()
    )
    strategy = (
        WatermarkStrategy.for_bounded_out_of_orderness(Duration.of_millis(OUT_OF_ORDERNESS_MS))
        .with_timestamp_assigner(ExtractEventTime())
        .with_idleness(Duration.of_seconds(3))
    )
    stream = env.from_source(source, strategy, "kafka-watermark-events")
    windowed = (
        stream.key_by(lambda v: json.loads(v)["sensor"])
        .window(TumblingEventTimeWindows.of(Time.seconds(WINDOW_SECONDS)))
        .allowed_lateness(ALLOWED_LATENESS_MS)
        .side_output_late_data(LATE_DATA_TAG)
        .process(PrintWindowResult())
    )
    windowed.get_side_output(LATE_DATA_TAG).process(PrintLateData())
    return env


def send(producer, sensor, offset_ms, base, tag=""):
    payload = json.dumps({"sensor": sensor, "ts": base + offset_ms, "off": offset_ms}).encode()
    producer.produce(TOPIC, key=sensor.encode(), value=payload)
    producer.flush(5)
    log(f"produced key={sensor} off={offset_ms} {tag}")


def scenario_late_but_allowed():
    """window0 у sensor-1 закрывается событием на ключе clock (не трогает elements
    sensor-1). Вскоре после закрытия шлём опоздавшее sensor-1 - ожидаем late firing."""
    log("=== сценарий 1: опоздание в пределах allowedLateness ===")
    from confluent_kafka import Producer

    def produce():
        time.sleep(2)
        producer = Producer({"bootstrap.servers": BOOTSTRAP_SERVERS})
        base = int(time.time() * 1000)
        send(producer, "sensor-1", 0, base)
        time.sleep(2)
        send(producer, "clock", 9000, base, tag="(двигает общий watermark)")
        time.sleep(6)
        # Дедлайн окна window0: end(5000) + allowedLateness(4000) = 9000 мс от начала.
        # Отправляем почти сразу после закрытия, пока watermark ещё далёк от дедлайна.
        send(producer, "sensor-1", 1000, base, tag="(опоздавшее, в пределах allowedLateness)")
        time.sleep(10)
        log("producer (сценарий 1) done")

    env = build_job("late_scenario_1")
    t = threading.Thread(target=produce, daemon=True)
    t.start()
    job_client = env.execute_async("watermark late firing demo")
    t.join()
    time.sleep(3)
    job_client.cancel()
    log("сценарий 1 завершён")


def scenario_too_late():
    """Та же схема, но опоздавшее событие отправляется после долгой паузы - к этому
    моменту watermark гарантированно ушёл за window_end + allowedLateness."""
    log("=== сценарий 2: опоздание за allowedLateness (side output) ===")
    from confluent_kafka import Producer

    def produce():
        time.sleep(2)
        producer = Producer({"bootstrap.servers": BOOTSTRAP_SERVERS})
        base = int(time.time() * 1000)
        send(producer, "sensor-1", 0, base)
        time.sleep(2)
        send(producer, "clock", 9000, base, tag="(двигает общий watermark)")
        time.sleep(10)
        log("--- долгая пауза, чтобы дедлайн window0 остался далеко позади ---")
        time.sleep(15)
        send(producer, "sensor-1", 1000, base, tag="(слишком поздно -> side output)")
        time.sleep(10)
        log("producer (сценарий 2) done")

    env = build_job("late_scenario_2")
    t = threading.Thread(target=produce, daemon=True)
    t.start()
    job_client = env.execute_async("watermark side output demo")
    t.join()
    time.sleep(3)
    job_client.cancel()
    log("сценарий 2 завершён")


if __name__ == "__main__":
    scenario_late_but_allowed()
    time.sleep(3)
    scenario_too_late()

Здесь честно про неудобный результат. Оба сценария на этом стенде корректно закрыли своё исходное окно, но повторно отправленное в уже закрытое окно событие sensor-1 ни разу не вызвало ни late firing, ни появления в side_output, хотя allowed_lateness и side_output_late_data настроены в точности по документированному API Flink. Похоже на ограничение Python-моста PyFlink 2.3.0 между ProcessFunction и WindowOperator, а не на ошибку конфигурации, потому что тот же код и тот же API на Java-стороне Flink в документации описаны без такой оговорки. Демо оставлено как есть, потому что показывает настоящее, а не подогнанное под красивую картинку поведение инструмента.

Заключение

Watermark переводит расплывчатое «событие могло опоздать» в конкретное управляемое число, ниже которого система готова считать окно закрытым. Три компонента, TimestampAssigner, WatermarkGenerator и WatermarkStrategy, решают эту задачу в Apache Flink, а Apache Beam описывает тот же принцип через понятия heuristic и perfect watermark и полноты входных данных. Живой прогон на PyFlink и Kafka показал обе стороны механизма, работающую пометку простаивающего источника через with_idleness и неудобную находку про live-демонстрацию allowedLateness, которая не всегда воспроизводится один в один с документацией на конкретной версии инструмента. Понимание watermark не заканчивается на этой странице, оно становится практическим только тогда, когда приходится вживую подбирать задержку bounded out-of-orderness под реальный профиль опозданий своих собственных источников.

Референсные ссылки

• Apache Flink, Generating Watermarks
• Apache Beam Programming Guide, Watermarks and Late Data
• Водяные знаки в заданиях Flink для потоковой обработки данных из Kafka