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

Для Storm потребуется поместить всю вашу логику в ваш код UDF, используя низкоуровневый Java API (я сомневаюсь, что Trindent полезен). Я никогда не работал с Samza и не могу помочь (или решить, какая система лучше подойдет для решения вашей проблемы).

Например, в Storm вы можете назначить временную метку каждому кортежу в Spout.nextTuple() и буферизовать все кортежи незавершенного платежа в Bolt в порядке убывания временной метки. Каждый раз, когда вызывается Bolt.execute(), вы можете сравнить отметку времени нового кортежа с начальным (т. е. самым старым кортежем) вашей очереди. Если входной кортеж имеет большую временную метку, чем head-T плюс X, вы знаете, что время ожидания вашего головного кортежа истекло, и вы можете поднять для него триггер.

Конечно, вам нужно выполнить fieldsGrouping(), чтобы убедиться, что все кортежи, принадлежащие одному и тому же платежу, обрабатываются одним и тем же экземпляром Bolt. Вам также может понадобиться несколько упорядочить входящие кортежи болтов по отметке времени или использовать более продвинутую логику тайм-аута для обработки неупорядоченных кортежей (в отношении увеличения отметок времени).

В зависимости от ваших требований к задержке и скорости входного потока вы также можете использовать «тиковые кортежи», чтобы инициировать сравнение головного кортежа с этим фиктивным кортежем тиков. Или, как еще более строгая реализация, проделайте всю эту логику прямо в Spout.next() (если вы знаете, что все кортежи платежа проходят через один и тот же экземпляр Spout).