Как отправить итоговый результат агрегации кафка-потоков из окна KTable с временным окном?

Что я хотел бы сделать, так это:

Использовать записи из темы номеров (Long's)
Совокупность (подсчет) значений для каждого окна 5 секунд
Отправьте результат агрегации FINAL в другую тему

Мой код выглядит следующим образом:

KStream<String, Long> longs = builder.stream(
        Serdes.String(), Serdes.Long(), "longs");

// In one ktable, count by key, on a five second tumbling window.
KTable<Windowed<String>, Long> longCounts = 
        longs.countByKey(TimeWindows.of("longCounts", 5000L));

// Finally, sink to the long-avgs topic.
longCounts.toStream((wk, v) -> wk.key())
        .to("long-counts");

Похоже, что все работает так, как ожидалось, но агрегаты отправляются в тему назначения для каждой входящей записи. Мой вопрос в том, как я могу отправить только окончательный результат агрегирования для каждого окна?

Ответ 1

В Kafka Streams не существует такой вещи, как "конечная агрегация". Окна постоянно открыты для обработки неупорядоченных записей, поступающих после истечения времени окончания окна. Однако окна не сохраняются вечно. Они сбрасываются после истечения срока их хранения. Никаких специальных действий в отношении того, когда окно отбрасывается, не существует.

См. документацию Confluent для получения дополнительной информации: http://docs.confluent.io/current/streams/

Таким образом, для каждого обновления агрегации создается запись результата (поскольку Kafka Streams также обновляет результат агрегации для записей, вышедших из строя). Ваш "конечный результат" будет самой последней записью результата (до того, как окно будет удалено). В зависимости от вашего варианта использования, ручное устранение дубликатов будет способом решения проблемы (с использованием API нижнего рычага, transform() или process())

Этот пост в блоге тоже может помочь: https://timothyrenner.github.io/engineering/2016/08/11/kafka-streams-not-looking-at-facebook.html

Еще одно сообщение в блоге, посвященное этой проблеме без знаков препинания: http://blog.inovatrend.com/2018/03/making-of-message-gateway-with-kafka.html

Обновление

С KIP-328 добавлен оператор KTable#suppress(), который позволит строго подавлять последовательные обновления и выдавать одну запись результата на окно; компромисс - увеличение задержки.

Ответ 2

Начиная с версии 2.1 Kafka Streams, вы можете добиться этого , используя suppress.

Вот пример из упомянутой документации Apache Kafka Streams, который отправляет предупреждение, когда у пользователя менее трех событий в час:

KGroupedStream<UserId, Event> grouped = ...;
grouped
  .windowedBy(TimeWindows.of(Duration.ofHours(1)).grace(ofMinutes(10)))
  .count()
  .suppress(Suppressed.untilWindowCloses(unbounded()))
  .filter((windowedUserId, count) -> count < 3)
  .toStream()
  .foreach((windowedUserId, count) -> sendAlert(windowedUserId.window(), windowedUserId.key(), count));

Как упоминалось в обновлении этого ответа, вы должны знать о компромиссе. Кроме того, обратите внимание, что suppress() основан на времени события.