Агрегировать записи в потоке на основе времени или маркера конца

Question

Рассмотрим понятие входного потока переплетенных записей, представляющих взаимодействие с пользователем (например, покупка продукта). Представьте, что мы получаем записи, которые указывают, что пользователь поместил товар в свою корзину. Через некоторое время они проводят проверку ... или ... оставляют свою корзину.

Таким образом, я получаю поток записей, таких как:

• Транзакция: 123, добавлен предмет A в корзину
• Транзакция: 123, добавлен товар B в корзину
• ...
• Транзакция: 123, проверенная корзина

Моя цель - вывести из конвейера агрегат транзакции. Например, учитывая вышеизложенное, я хочу вывести:

Transaction 123, Items A, B, ... Sale completed

или если выезд не производится в течение 24 часов после последнего события:

Transaction 123, Items A, B, ... Sale abandoned

... и вот где я застрял. Я чувствую, что есть какой-то способ подумать об этой истории с точки зрения конвейера Apache Beam, но я боюсь, что не знаю, с чего начать. Я думаю, что я как-то хочу, чтобы окно записей как по транзакции, так и по завершению, и отправлять партию для обработки только тогда, когда получено либо end записи транзакции, либо какой-то временной интервал с момента последней записи .

Answer 1

Маркеры окон на основе данных имеют внутреннее предположение о порядке данных, которое не поддерживается Beam.В приведенном выше сценарии предполагается, что событие извлечения наступит после всех событий добавления в корзину.

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

PCollection-RawEvents: читать необработанные события

PCollection-1: PCollection-RawEvents -> Применить 24 часа SessionWindow ко всем событиям.

PCollection-Checkout: PCollection-1 -> Нажмите все элементы для ввода ключа в BagState.Считайте состояние и опубликуйте событие Transaction 123, Items A, B, ... Sale completed, когда вы получите событие извлечения Transaction: 123, Checked out basket.

PCollection-Abandon: PCollection-1 -> GroupByKey -> Publish Transaction 123, Items A, B, ... Sale abandoned, если Transaction: 123, Checked out basket отсутствует.

PCollection-Unified: Свести (PCollection-Checkout, PCollection-Abandon)