dropDueToLateness и dropDueToClosedWindow в потоке данных

Question

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

Window.into(FixedWindows.of(Duration.standardMinutes(15)))
            .triggering(AfterWatermark
                .pastEndOfWindow())
            .withAllowedLateness(Duration.ZERO)
            .discardingFiredPanes()

Моя первоначальная мысль об объекте Window, который срабатывает каждые 15минут предыдущий. Однако такой подход дает мне droppedDueToLateness. Чтобы решить эту проблему, я подумал: «Хорошо, давайте тогда увеличим допустимое опоздание!».

Изменил следующую строку .withAllowedLateness(Duration.standardMinutes(60)), чтобы поздние события могли все же попасть в окно.

Однакопосле внесения этого изменения отброшенные события droppedDueToClosedWindow снова возвращаются, которых я пытался избежать в моих первоначальных реализациях! (Не включен в этот вопрос, но на самом деле не связан с вопросом).

Сообщения приходят из очереди PubSub с событием временной метки из самой полезной нагрузки сообщения (атрибут внутри объекта JSON, а не метка времени из PubSub).

Есть какие-нибудь подсказки, почему это может происходить? Должен ли я просто увеличить ресурсы или все же немного подправить объект Window?

Answer 1

Я получил много dropDueToClosedWindow и переключил свое окно на очень простое

Window.into(FixedWindows.of(windowDuration)))

Затем они ушли.

Просмотр Шаблоны потока данных Google они используютОконный один из двух способов. Очень простой способ, указанный выше. И в DLPTextToBigQueryStreaming они делают следующее, которое похоже на то, что вы делали. У меня больше успеха с первым подходом.

                Window.<KV<String, ReadableFile>>into(FixedWindows.of(WINDOW_INTERVAL))
                .triggering(
                    AfterProcessingTime.pastFirstElementInPane().plusDelayOf(Duration.ZERO))
                .discardingFiredPanes()
                .withAllowedLateness(Duration.ZERO))

Answer 2

Не могли бы вы поделиться кодом, где вы читаете данные с PubSubIO? Вы использовали метод withTimestampAttribute?