Я бы предложил KSQL, который представляет собой механизм потокового SQL, который позволяет обрабатывать данные в реальном времени против Апач Кафка. Он также обеспечивает хорошую функциональность для оконного агрегирования и т. д.

Существует 3 способа определить Windows в KSQL:

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

В вашем контексте вы можете использовать KSQL для запроса и объединения интересующих тем с помощью Оконные соединения. Например,

SELECT t1.id, ...
  FROM topic_1 t1
  INNER JOIN topic_2 t2
    WITHIN 1 HOURS
    ON t1.id = t2.id;

Некоторые предложения -

1. Обработка задержки на стороне производителя. Убедитесь, что все три производителя всегда синхронно отправляют данные в темы Kafka, используя batch.size и linger.ms. например. если для linger.ms установлено значение 1000, все сообщения будут отправляться в Kafka в течение 1 секунды.

2. Обрабатывать задержку на стороне потребителя. Принимая во внимание любой механизм потоковой передачи на стороне потребителя (будь то Kafka-stream, spark-stream, Flink), предоставляет функциональные возможности Windows для объединения/объединения потоковых данных на основе ключей с учетом функции окна с задержкой.

Проверьте это - окна Flink для справки, как выбрать правильный тип окна ссылка

Чтобы справиться с этим сценарием, источники данных должны предоставить потребителю некоторый механизм, позволяющий понять, что все соответствующие данные получены. Самое простое решение — опубликовать пакет из источника данных с идентификатором пакета (Guid) в той или иной форме. Затем потребители могут подождать, пока не появится идентификатор следующей партии, отмечающий конец предыдущей партии. Этот подход предполагает, что источники не будут пропускать пакет, иначе они будут постоянно смещены. Не существует алгоритма для обнаружения этого, но у вас могут быть некоторые поля в данных, которые показывают неоднородность и позволяют вам повторно выравнивать данные.

Более слабая версия этого подхода состоит в том, чтобы либо просто подождать x секунд и предположить, что все источники преуспевают в течение этого большого количества времени, либо посмотреть на некоторую форму меток времени (логические или настенные часы), чтобы определить, что источник перешел к следующему временному окну. неявно показывая завершение последнего окна.

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

1. Используйте синхронизатор сетевого времени на всех узлах производителя/потребителя.
2. Используйте контрольное сообщение от производителей каждые x единиц времени с фиксированным временем начала. Например: сообщения отправляются каждые две минуты в начале минуты.
3. Создайте предикторы для задержки сообщения производителя. используйте сообщения сердцебиения, чтобы вычислить это.

С помощью этих примитивов мы должны иметь возможность выравнивать события временных рядов, учитывая отклонения во времени из-за сетевых задержек.

На стороне механизма логического вывода расширьте окна на уровне каждого производителя, чтобы синхронизировать события между производителями.