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

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

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

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

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

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

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

Есть 3 способа определить Окна в KSQL:

hopping windows, tumbling windows, and session windows. Hopping and tumbling windows are time windows, because they're defined by fixed durations they you specify. Session windows are dynamically sized based on incoming data and defined by periods of activity separated by gaps of inactivity.

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

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

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

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

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

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