Обнаружение выбросов и пропусков данных в Storm

Реализовать в инфраструктуре Storm (потоковая обработка данных). На Java.

На вход поступает большой файл формата .csv с двумя колонками данных: TimeStamp и Value.

Пример:

Timestamp Value

01.01.2010 0:00 4

01.01.2010 0:15 32

01.01.2010 0:30 12

01.01.2010 0:45 20

...

Сделать конвейер обработки данных с генерацией следующих топологий:

1) выявление выбросов (есть пороговое значение, если value больше, то происходит event и запись в БД)

2) выявление пропусков:

2.1) если текущее значение value не инициилизированно, то event и запись в БД

2.2) хранится значение предыдущего TimeStamp. считается разница между текущим и предыдущим. если модуль больше 15 минут, то event и запись в БД.