Анализ потоков Kafka для обнаружения аномалий ⇐ JAVA

[Anonymous]

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

События
В теме Kafka есть поток входящих событий:

Код: Выделить всё

=====> E5,E4,E3,E2,E1,E0 =====>

Каждое событие представлено доменным классом MyEvent.

Код: Выделить всё

public class MyEvent{

private String eventKey; // which is same as the kafka message key
...........
........... //business fields
private LocalDateTime timestamp;

private long runningTotal; // Important Field
}

И существует постоянный поток событий, которые публикуются во входной теме темы.
The RunningTotal code> увеличивается со временем, и это значение публикуется внешней системой.
На этом фоне мне приходится обнаруживать аномалии в потоке событий:

Временные окна
Сценарий использования заключается в обнаружении любого внезапного всплеска в поле RunningTotal с тем же eventKey.
И этот всплеск происходит в течение определенного периода времени, скажем, 60 секунд (настраивается) и пороговое значение(настраивается, пример: 500)
Пример:

• Код: Выделить всё

  t0

  : eventKey=key1, RunningTotal = 10

• Код: Выделить всё

  t60(seconds)

  : eventKey=key1, RunningTotal = 50 // ок.. потому что внезапного всплеска нет

• Код: Выделить всё

  t120(seconds)

  : eventKey=key1, RunningTotal = 150000 // не ок.. потому что произошел внезапный всплеск

Я думаю, что это вариант использования Kafka Streams с SlidingWindows, .filter() и .aggregate().
Но я я не могу чтобы понять, как реализовать это в Java/Spring Boot.
Буду признателен за любую помощь.

Код: Выделить всё

KStream streamIn = streamsBuilder.stream(TOPIC_IN, Consumed.with(Serdes.String(), new JsonSerde(MyEvent.class)));

streamIn.

//.....do some magic here.....

.to("topic-out")

Выходные данные должны быть опубликованы для обсуждения.


Подробнее здесь: https://stackoverflow.com/questions/793 ... -detection