Как задать временные окна в потоковой обработке данных?

В Apache Kafka Streams и Kafka Connect с KSQL временные окна определяются для агрегации событий в рамках заданных временных интервалов. Основные типы:

• Tumbling Window (фиксированные): Непересекающиеся окна фиксированной длины (например, каждые 5 минут).
• Hopping Window (скользящие): Пересекающиеся окна с заданной длиной и интервалом продвижения (например, окно 5 минут, обновляемое каждую минуту).
• Session Window (сессионные): Динамические окна, которые расширяются при поступлении событий и закрываются после периода неактивности.

Пример на KSQL для Tumbling Window (агрегация за 30 секунд):

CREATE TABLE page_views_per_region AS
  SELECT regionid,
         COUNT(*) AS view_count
  FROM page_views
  WINDOW TUMBLING (SIZE 30 SECONDS)
  GROUP BY regionid
  EMIT CHANGES;

Пример на Kafka Streams API (Java) для Hopping Window:

KGroupedStream<String, String> groupedStream = ...;

KTable<Windowed<String>, Long> counts = groupedStream
    .windowedBy(TimeWindows.of(Duration.ofMinutes(5)).advanceBy(Duration.ofMinutes(1))) // Окно 5 мин, шаг 1 мин
    .count(Materialized.as("counts-store"));

Ключевые аспекты:

• Обработка по времени события (event-time): Требует, чтобы события содержали временную метку. Для этого используется TimestampExtractor.
• Гранулярность удержания (retention period): Настраивается для State Stores, чтобы старые окна могли быть удалены и освободить память.
• Обработка опоздавших событий (late arrival): Настраивается через параметр grace period в окне, чтобы позволить обработать события, пришедшие с задержкой.