Окно луча Apache: учитывайте запоздалые данные, но излучайте только одну панель

Question

Я хотел бы создать одну панель, когда водяной знак достигает x минут после конца окна. Это позволяет мне обрабатывать некоторые запоздалые данные, но при этом генерировать только одну панель. В настоящее время я работаю в Java.

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

В настоящее время у меня есть код, подобный этому:

.triggering(
    // This is going to emit the pane, but I don't want emit the pane yet!                                  
    AfterWatermark.pastEndOfWindow()

    // This is going to emit panes each time I receive late data, however 
    // I would like to only emit one pane at the end of the allowedLateness
).withAllowedLateness(allowedLateness).accumulatingFiredPanes())

Если все еще есть путаница, я хотел бы создать только одну панель, когда водяной знак проходит allowedLateness.

Answer 1

Спасибо, Гиллем, в конце концов, я использовал ваш ответ, чтобы найти очень полезную ссылку с множеством примеров Apache Beam.Из этого я пришел к следующему решению:

 // We first specify to never emit any panes
 .triggering(Never.ever())

 // We then specify to fire always when closing the window. This will emit a
 // single final pane at the end of allowedLateness
 .withAllowedLateness(allowedLateness, Window.ClosingBehavior.FIRE_ALWAYS)
 .discardingFiredPanes())

Answer 2

Сначала я бы установил Window.ClosingBehavior на FIRE_ALWAYS.Таким образом, когда окно постоянно закрыто, оно отправит окончательную панель (даже если с момента последней панели не было поздних записей) с PaneInfo.isLast, установленным на true.

Затем я перешел бы ко второму варианту:

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

Но отбрасывая нижестоящие панели, которые не являются окончательными, что-то вроде:

public void processElement(ProcessContext c) {
    if (c.pane().isLast) {
        c.output(c.element());
    }
}