Apache Beam - Дельта между окнами

Question

Извинения , пытаясь быть кратким и ясным, мое предыдущее описание моего вопроса превратилось в частный случай общего случая, который я пытаюсь разгадать.

Новое описание

Я пытаюсь сравнить последнее переданное значение функции агрегирования (скажем, Sum ()) с каждым элементом, который я собираю в текущем окне.

Стоит отметить, что идеальное (я думаю) решение будет включать

• Элемент T2 (из t-1), используемый во время = t, - это элемент, созданный в предыдущем окне.

Я играл с несколькими идеями / экспериментами, но я изо всех сил пытаюсь найти способ сделать это способом, элегантным и «чутким» к вычислительной модели Бима (которую я до сих пор пытаюсь полностью воплотить в жизнь после многих статья / блог / документация и книга:)

Боковые входы кажутся громоздкими, потому что, похоже, мне нужно сдвинуть метку времени испущенной Агрегации 5M @ T-1 в окно 5M @ T, чтобы выровнять ее с текущим окном 5M

Пытаясь сделать это с побочными вводами (как я их понимаю), я получил некоторый неприятный код, который был довольно "циклически ссылочным", но не элегантным рекурсивным способом:)

Любая помощь в правильном направлении будет принята с благодарностью.

Edit: Модифицированная схема и улучшенное описание для более наглядного отображения:

• Намерение использовать испущенный T2 (из t-1) для вычисления T2 при t
• что требуемый T2 (из t-1), используемый для вычисления T2, - это тот, который имеет правильный ключ

Answer 1

Вместо того, чтобы изменять временную метку записей, которые материализуются так, чтобы они появлялись в текущем окне, вы должны предоставить отображение окна fn, которое просто отображает текущее окно на предыдущее. Вы захотите создать пользовательскую WindowFn , которая реализует поведение сопоставления окон, которое вы хотите, уделяя особое внимание переопределению функции getDefaultWindowMappingFn .

Ваш конвейер будет выглядеть так:

PCollection<T> mySource = /* data */
PCollectionView<SumT> view = mySource
    .apply(Window.into(myCustomWindowFnWithNewWindowMappingFn))
    .apply(Combine.globally(myCombiner).asSingletonView());

mySource.apply(ParDo.of(/* DoFn that consumes side input */).withSideInputs(view));

Обратите особое внимание на значение по умолчанию, которое выдаст объединитель, так как это будет значение по умолчанию, когда в представление не было передано никаких данных. Кроме того, самый простой способ написать собственную пользовательскую оконную функцию - скопировать существующую.